diff --git a/importer/balanced/balanced_test.go b/importer/balanced/balanced_test.go index 5968d6f65..947867cb6 100644 --- a/importer/balanced/balanced_test.go +++ b/importer/balanced/balanced_test.go @@ -22,15 +22,12 @@ import ( // TODO: extract these tests and more as a generic layout test suite func buildTestDag(ds dag.DAGService, spl chunk.Splitter) (*dag.Node, error) { - // Start the splitter - blkch, errs := chunk.Chan(spl) - dbp := h.DagBuilderParams{ Dagserv: ds, Maxlinks: h.DefaultLinksPerBlock, } - return BalancedLayout(dbp.New(blkch, errs)) + return BalancedLayout(dbp.New(spl)) } func getTestDag(t *testing.T, ds dag.DAGService, size int64, blksize int64) (*dag.Node, []byte) { diff --git a/importer/helpers/dagbuilder.go b/importer/helpers/dagbuilder.go index 1d9f0bd10..bec8d4106 100644 --- a/importer/helpers/dagbuilder.go +++ b/importer/helpers/dagbuilder.go @@ -1,6 +1,7 @@ package helpers import ( + "github.com/ipfs/go-ipfs/importer/chunk" dag "github.com/ipfs/go-ipfs/merkledag" ) @@ -8,8 +9,7 @@ import ( // efficiently create unixfs dag trees type DagBuilderHelper struct { dserv dag.DAGService - in <-chan []byte - errs <-chan error + spl chunk.Splitter recvdErr error nextData []byte // the next item to return. maxlinks int @@ -24,45 +24,35 @@ type DagBuilderParams struct { Dagserv dag.DAGService } -// Generate a new DagBuilderHelper from the given params, using 'in' as a -// data source -func (dbp *DagBuilderParams) New(in <-chan []byte, errs <-chan error) *DagBuilderHelper { +// Generate a new DagBuilderHelper from the given params, which data source comes +// from chunks object +func (dbp *DagBuilderParams) New(spl chunk.Splitter) *DagBuilderHelper { return &DagBuilderHelper{ dserv: dbp.Dagserv, - in: in, - errs: errs, + spl: spl, maxlinks: dbp.Maxlinks, batch: dbp.Dagserv.Batch(), } } -// prepareNext consumes the next item from the channel and puts it +// prepareNext consumes the next item from the splitter and puts it // in the nextData field. it is idempotent-- if nextData is full // it will do nothing. -// -// i realized that building the dag becomes _a lot_ easier if we can -// "peek" the "are done yet?" (i.e. not consume it from the channel) func (db *DagBuilderHelper) prepareNext() { - if db.in == nil { - // if our input is nil, there is "nothing to do". we're done. - // as if there was no data at all. (a sort of zero-value) - return - } - - // if we already have data waiting to be consumed, we're ready. + // if we already have data waiting to be consumed, we're ready if db.nextData != nil { return } - // if it's closed, nextData will be correctly set to nil, signaling - // that we're done consuming from the channel. - db.nextData = <-db.in + // TODO: handle err (which wasn't handled either when the splitter was channeled) + db.nextData, _ = db.spl.NextBytes() } // Done returns whether or not we're done consuming the incoming data. func (db *DagBuilderHelper) Done() bool { // ensure we have an accurate perspective on data // as `done` this may be called before `next`. + //db.prepareNext() // idempotent db.prepareNext() // idempotent return db.nextData == nil } diff --git a/importer/importer.go b/importer/importer.go index b16b5b05b..92faddd7a 100644 --- a/importer/importer.go +++ b/importer/importer.go @@ -39,25 +39,19 @@ func BuildDagFromFile(fpath string, ds dag.DAGService) (*dag.Node, error) { } func BuildDagFromReader(ds dag.DAGService, spl chunk.Splitter) (*dag.Node, error) { - // Start the splitter - blkch, errch := chunk.Chan(spl) - dbp := h.DagBuilderParams{ Dagserv: ds, Maxlinks: h.DefaultLinksPerBlock, } - return bal.BalancedLayout(dbp.New(blkch, errch)) + return bal.BalancedLayout(dbp.New(spl)) } func BuildTrickleDagFromReader(ds dag.DAGService, spl chunk.Splitter) (*dag.Node, error) { - // Start the splitter - blkch, errch := chunk.Chan(spl) - dbp := h.DagBuilderParams{ Dagserv: ds, Maxlinks: h.DefaultLinksPerBlock, } - return trickle.TrickleLayout(dbp.New(blkch, errch)) + return trickle.TrickleLayout(dbp.New(spl)) } diff --git a/importer/trickle/trickle_test.go b/importer/trickle/trickle_test.go index 2cd98ec97..6b1e0f346 100644 --- a/importer/trickle/trickle_test.go +++ b/importer/trickle/trickle_test.go @@ -21,15 +21,12 @@ import ( ) func buildTestDag(ds merkledag.DAGService, spl chunk.Splitter) (*merkledag.Node, error) { - // Start the splitter - blkch, errs := chunk.Chan(spl) - dbp := h.DagBuilderParams{ Dagserv: ds, Maxlinks: h.DefaultLinksPerBlock, } - nd, err := TrickleLayout(dbp.New(blkch, errs)) + nd, err := TrickleLayout(dbp.New(spl)) if err != nil { return nil, err } @@ -441,10 +438,9 @@ func TestAppend(t *testing.T) { } r := bytes.NewReader(should[nbytes/2:]) - blks, errs := chunk.Chan(chunk.NewSizeSplitter(r, 500)) ctx := context.Background() - nnode, err := TrickleAppend(ctx, nd, dbp.New(blks, errs)) + nnode, err := TrickleAppend(ctx, nd, dbp.New(chunk.NewSizeSplitter(r, 500))) if err != nil { t.Fatal(err) } @@ -494,9 +490,8 @@ func TestMultipleAppends(t *testing.T) { ctx := context.Background() for i := 0; i < len(should); i++ { - blks, errs := chunk.Chan(spl(bytes.NewReader(should[i : i+1]))) - nnode, err := TrickleAppend(ctx, nd, dbp.New(blks, errs)) + nnode, err := TrickleAppend(ctx, nd, dbp.New(spl(bytes.NewReader(should[i:i+1])))) if err != nil { t.Fatal(err) } @@ -538,17 +533,13 @@ func TestAppendSingleBytesToEmpty(t *testing.T) { spl := chunk.SizeSplitterGen(500) - blks, errs := chunk.Chan(spl(bytes.NewReader(data[:1]))) - ctx := context.Background() - nnode, err := TrickleAppend(ctx, nd, dbp.New(blks, errs)) + nnode, err := TrickleAppend(ctx, nd, dbp.New(spl(bytes.NewReader(data[:1])))) if err != nil { t.Fatal(err) } - blks, errs = chunk.Chan(spl(bytes.NewReader(data[1:]))) - - nnode, err = TrickleAppend(ctx, nnode, dbp.New(blks, errs)) + nnode, err = TrickleAppend(ctx, nnode, dbp.New(spl(bytes.NewReader(data[1:])))) if err != nil { t.Fatal(err) } diff --git a/unixfs/mod/dagmodifier.go b/unixfs/mod/dagmodifier.go index aa4de8caf..197e330a9 100644 --- a/unixfs/mod/dagmodifier.go +++ b/unixfs/mod/dagmodifier.go @@ -103,8 +103,7 @@ func (zr zeroReader) Read(b []byte) (int, error) { func (dm *DagModifier) expandSparse(size int64) error { r := io.LimitReader(zeroReader{}, size) spl := chunk.NewSizeSplitter(r, 4096) - blks, errs := chunk.Chan(spl) - nnode, err := dm.appendData(dm.curNode, blks, errs) + nnode, err := dm.appendData(dm.curNode, spl) if err != nil { return err } @@ -191,8 +190,7 @@ func (dm *DagModifier) Sync() error { // need to write past end of current dag if !done { - blks, errs := chunk.Chan(dm.splitter(dm.wrBuf)) - nd, err = dm.appendData(dm.curNode, blks, errs) + nd, err = dm.appendData(dm.curNode, dm.splitter(dm.wrBuf)) if err != nil { return err } @@ -286,13 +284,13 @@ func (dm *DagModifier) modifyDag(node *mdag.Node, offset uint64, data io.Reader) } // appendData appends the blocks from the given chan to the end of this dag -func (dm *DagModifier) appendData(node *mdag.Node, blks <-chan []byte, errs <-chan error) (*mdag.Node, error) { +func (dm *DagModifier) appendData(node *mdag.Node, spl chunk.Splitter) (*mdag.Node, error) { dbp := &help.DagBuilderParams{ Dagserv: dm.dagserv, Maxlinks: help.DefaultLinksPerBlock, } - return trickle.TrickleAppend(dm.ctx, node, dbp.New(blks, errs)) + return trickle.TrickleAppend(dm.ctx, node, dbp.New(spl)) } // Read data from this dag starting at the current offset