merkledag: switch to new dag interface

Also:

* Update the blockstore/blockservice methods to match.
* Construct a new temporary offline dag instead of having a
  GetOfflineLinkService method.

License: MIT
Signed-off-by: Steven Allen <steven@stebalien.com>

blockservice.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 
 	"github.com/ipfs/go-ipfs/blocks/blockstore"
 	exchange "github.com/ipfs/go-ipfs/exchange"
@@ -21,10 +22,27 @@ var log = logging.Logger("blockservice")
 
 var ErrNotFound = errors.New("blockservice: key not found")
 
+type BlockGetter interface {
+	// GetBlock gets the requested block.
+	GetBlock(ctx context.Context, c *cid.Cid) (blocks.Block, error)
+
+	// GetBlocks does a batch request for the given cids, returning blocks as
+	// they are found, in no particular order.
+	//
+	// It may not be able to find all requested blocks (or the context may
+	// be canceled). In that case, it will close the channel early. It is up
+	// to the consumer to detect this situation and keep track which blocks
+	// it has received and which it hasn't.
+	GetBlocks(ctx context.Context, ks []*cid.Cid) <-chan blocks.Block
+}
+
 // BlockService is a hybrid block datastore. It stores data in a local
 // datastore and may retrieve data from a remote Exchange.
 // It uses an internal `datastore.Datastore` instance to store values.
 type BlockService interface {
+	io.Closer
+	BlockGetter
+
 	// Blockstore returns a reference to the underlying blockstore
 	Blockstore() blockstore.Blockstore
 
@@ -32,20 +50,14 @@ type BlockService interface {
 	Exchange() exchange.Interface
 
 	// AddBlock puts a given block to the underlying datastore
-	AddBlock(o blocks.Block) (*cid.Cid, error)
+	AddBlock(o blocks.Block) error
 
 	// AddBlocks adds a slice of blocks at the same time using batching
 	// capabilities of the underlying datastore whenever possible.
-	AddBlocks(bs []blocks.Block) ([]*cid.Cid, error)
-
-	GetBlock(ctx context.Context, c *cid.Cid) (blocks.Block, error)
-	DeleteBlock(o blocks.Block) error
+	AddBlocks(bs []blocks.Block) error
 
-	// GetBlocks does a batch request for the given cids, returning blocks as
-	// they are found, in no particular order.
-	GetBlocks(ctx context.Context, ks []*cid.Cid) <-chan blocks.Block
-
-	Close() error
+	// DeleteBlock deletes the given block from the blockservice.
+	DeleteBlock(o *cid.Cid) error
 }
 
 type blockService struct {
@@ -110,38 +122,34 @@ func NewSession(ctx context.Context, bs BlockService) *Session {
 
 // AddBlock adds a particular block to the service, Putting it into the datastore.
 // TODO pass a context into this if the remote.HasBlock is going to remain here.
-func (s *blockService) AddBlock(o blocks.Block) (*cid.Cid, error) {
+func (s *blockService) AddBlock(o blocks.Block) error {
 	c := o.Cid()
 	if s.checkFirst {
-		has, err := s.blockstore.Has(c)
-		if err != nil {
-			return nil, err
-		}
-
-		if has {
-			return c, nil
+		if has, err := s.blockstore.Has(c); has || err != nil {
+			return err
 		}
 	}
 
-	err := s.blockstore.Put(o)
-	if err != nil {
-		return nil, err
+	if err := s.blockstore.Put(o); err != nil {
+		return err
 	}
 
 	if err := s.exchange.HasBlock(o); err != nil {
-		return nil, errors.New("blockservice is closed")
+		// TODO(stebalien): really an error?
+		return errors.New("blockservice is closed")
 	}
 
-	return c, nil
+	return nil
 }
 
-func (s *blockService) AddBlocks(bs []blocks.Block) ([]*cid.Cid, error) {
+func (s *blockService) AddBlocks(bs []blocks.Block) error {
 	var toput []blocks.Block
 	if s.checkFirst {
+		toput = make([]blocks.Block, 0, len(bs))
 		for _, b := range bs {
 			has, err := s.blockstore.Has(b.Cid())
 			if err != nil {
-				return nil, err
+				return err
 			}
 			if !has {
 				toput = append(toput, b)
@@ -153,18 +161,16 @@ func (s *blockService) AddBlocks(bs []blocks.Block) ([]*cid.Cid, error) {
 
 	err := s.blockstore.PutMany(toput)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
-	var ks []*cid.Cid
 	for _, o := range toput {
 		if err := s.exchange.HasBlock(o); err != nil {
-			return nil, fmt.Errorf("blockservice is closed (%s)", err)
+			// TODO(stebalien): Should this really *return*?
+			return fmt.Errorf("blockservice is closed (%s)", err)
 		}
-
-		ks = append(ks, o.Cid())
 	}
-	return ks, nil
+	return nil
 }
 
 // GetBlock retrieves a particular block from the service,
@@ -256,8 +262,8 @@ func getBlocks(ctx context.Context, ks []*cid.Cid, bs blockstore.Blockstore, f e
 }
 
 // DeleteBlock deletes a block in the blockservice from the datastore
-func (s *blockService) DeleteBlock(o blocks.Block) error {
-	return s.blockstore.DeleteBlock(o.Cid())
+func (s *blockService) DeleteBlock(c *cid.Cid) error {
+	return s.blockstore.DeleteBlock(c)
 }
 
 func (s *blockService) Close() error {

test/blocks_test.go

@@ -33,16 +33,12 @@ func TestBlocks(t *testing.T) {
 		t.Error("Block key and data multihash key not equal")
 	}
 
-	k, err := bs.AddBlock(o)
+	err := bs.AddBlock(o)
 	if err != nil {
 		t.Error("failed to add block to BlockService", err)
 		return
 	}
 
-	if !k.Equals(o.Cid()) {
-		t.Error("returned key is not equal to block key", err)
-	}
-
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
 	defer cancel()
 	b2, err := bs.GetBlock(ctx, o.Cid())