feat: make the CoreAPI expose a streaming pin interface

pin.go

@@ -14,6 +14,9 @@ type Pin interface {
 
 	// Type of the pin
 	Type() string
+
+	// if not nil, an error happened. Everything else should be ignored.
+	Err() error
 }
 
 // PinStatus holds information about pin health
@@ -41,7 +44,7 @@ type PinAPI interface {
 	Add(context.Context, path.Path, ...options.PinAddOption) error
 
 	// Ls returns list of pinned objects on this node
-	Ls(context.Context, ...options.PinLsOption) ([]Pin, error)
+	Ls(context.Context, ...options.PinLsOption) (<-chan Pin, error)
 
 	// Rm removes pin for object specified by the path
 	Rm(context.Context, path.Path, ...options.PinRmOption) error

tests/block.go

@@ -1,18 +1,34 @@
 package tests
 
 import (
+	"bytes"
 	"context"
-	"github.com/ipfs/interface-go-ipfs-core/path"
+	"io"
 	"io/ioutil"
 	"strings"
 	"testing"
 
 	coreiface "github.com/ipfs/interface-go-ipfs-core"
 	opt "github.com/ipfs/interface-go-ipfs-core/options"
+	"github.com/ipfs/interface-go-ipfs-core/path"
 
 	mh "github.com/multiformats/go-multihash"
 )
 
+var (
+	pbCid    = "QmZULkCELmmk5XNfCgTnCyFgAVxBRBXyDHGGMVoLFLiXEN"
+	cborCid  = "bafyreicnga62zhxnmnlt6ymq5hcbsg7gdhqdu6z4ehu3wpjhvqnflfy6nm"
+	cborKCid = "bafyr2qgsohbwdlk7ajmmbb4lhoytmest4wdbe5xnexfvtxeatuyqqmwv3fgxp3pmhpc27gwey2cct56gloqefoqwcf3yqiqzsaqb7p4jefhcw"
+)
+
+func pbBlock() io.Reader {
+	return bytes.NewReader([]byte{10, 12, 8, 2, 18, 6, 104, 101, 108, 108, 111, 10, 24, 6})
+}
+
+func cborBlock() io.Reader {
+	return bytes.NewReader([]byte{101, 72, 101, 108, 108, 111})
+}
+
 func (tp *TestSuite) TestBlock(t *testing.T) {
 	tp.hasApi(t, func(api coreiface.CoreAPI) error {
 		if api.Block() == nil {
@@ -38,12 +54,12 @@ func (tp *TestSuite) TestBlockPut(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	res, err := api.Block().Put(ctx, strings.NewReader(`Hello`))
+	res, err := api.Block().Put(ctx, pbBlock())
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	if res.Path().Cid().String() != "QmPyo15ynbVrSTVdJL9th7JysHaAbXt9dM9tXk1bMHbRtk" {
+	if res.Path().Cid().String() != pbCid {
 		t.Errorf("got wrong cid: %s", res.Path().Cid().String())
 	}
 }
@@ -56,12 +72,12 @@ func (tp *TestSuite) TestBlockPutFormat(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	res, err := api.Block().Put(ctx, strings.NewReader(`Hello`), opt.Block.Format("cbor"))
+	res, err := api.Block().Put(ctx, cborBlock(), opt.Block.Format("cbor"))
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	if res.Path().Cid().String() != "bafyreiayl6g3gitr7ys7kyng7sjywlrgimdoymco3jiyab6rozecmoazne" {
+	if res.Path().Cid().String() != cborCid {
 		t.Errorf("got wrong cid: %s", res.Path().Cid().String())
 	}
 }
@@ -74,12 +90,17 @@ func (tp *TestSuite) TestBlockPutHash(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	res, err := api.Block().Put(ctx, strings.NewReader(`Hello`), opt.Block.Hash(mh.KECCAK_512, -1))
+	res, err := api.Block().Put(
+		ctx,
+		cborBlock(),
+		opt.Block.Hash(mh.KECCAK_512, -1),
+		opt.Block.Format("cbor"),
+	)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	if res.Path().Cid().String() != "bafyb2qgdh7w6dcq24u65xbtdoehyavegnpvxcqce7ttvs6ielgmwdfxrahmu37d33atik57x5y6s7d7qz32aasuwgirh3ocn6ywswqdifvu6e" {
+	if res.Path().Cid().String() != cborKCid {
 		t.Errorf("got wrong cid: %s", res.Path().Cid().String())
 	}
 }
@@ -92,7 +113,7 @@ func (tp *TestSuite) TestBlockGet(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	res, err := api.Block().Put(ctx, strings.NewReader(`Hello`), opt.Block.Hash(mh.KECCAK_512, -1))
+	res, err := api.Block().Put(ctx, strings.NewReader(`Hello`), opt.Block.Format("raw"))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -130,7 +151,7 @@ func (tp *TestSuite) TestBlockRm(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	res, err := api.Block().Put(ctx, strings.NewReader(`Hello`))
+	res, err := api.Block().Put(ctx, strings.NewReader(`Hello`), opt.Block.Format("raw"))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -184,7 +205,7 @@ func (tp *TestSuite) TestBlockStat(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	res, err := api.Block().Put(ctx, strings.NewReader(`Hello`))
+	res, err := api.Block().Put(ctx, strings.NewReader(`Hello`), opt.Block.Format("raw"))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -211,7 +232,7 @@ func (tp *TestSuite) TestBlockPin(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	_, err = api.Block().Put(ctx, strings.NewReader(`Hello`))
+	_, err = api.Block().Put(ctx, strings.NewReader(`Hello`), opt.Block.Format("raw"))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -220,14 +241,19 @@ func (tp *TestSuite) TestBlockPin(t *testing.T) {
 		t.Fatal("expected 0 pins")
 	}
 
-	res, err := api.Block().Put(ctx, strings.NewReader(`Hello`), opt.Block.Pin(true))
+	res, err := api.Block().Put(
+		ctx,
+		strings.NewReader(`Hello`),
+		opt.Block.Pin(true),
+		opt.Block.Format("raw"),
+	)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	pins, err := api.Pin().Ls(ctx)
+	pins, err := accPins(api.Pin().Ls(ctx))
 	if err != nil {
-		return
+		t.Fatal(err)
 	}
 	if len(pins) != 1 {
 		t.Fatal("expected 1 pin")

tests/pin.go

@@ -67,7 +67,7 @@ func (tp *TestSuite) TestPinSimple(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	list, err := api.Pin().Ls(ctx)
+	list, err := accPins(api.Pin().Ls(ctx))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -89,7 +89,7 @@ func (tp *TestSuite) TestPinSimple(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	list, err = api.Pin().Ls(ctx)
+	list, err = accPins(api.Pin().Ls(ctx))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -141,7 +141,7 @@ func (tp *TestSuite) TestPinRecursive(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	list, err := api.Pin().Ls(ctx)
+	list, err := accPins(api.Pin().Ls(ctx))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -150,7 +150,7 @@ func (tp *TestSuite) TestPinRecursive(t *testing.T) {
 		t.Errorf("unexpected pin list len: %d", len(list))
 	}
 
-	list, err = api.Pin().Ls(ctx, opt.Pin.Type.Direct())
+	list, err = accPins(api.Pin().Ls(ctx, opt.Pin.Type.Direct()))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -160,10 +160,10 @@ func (tp *TestSuite) TestPinRecursive(t *testing.T) {
 	}
 
 	if list[0].Path().String() != path.IpldPath(nd3.Cid()).String() {
-		t.Errorf("unexpected path, %s != %s", list[0].Path().String(), path.IpfsPath(nd2.Cid()).String())
+		t.Errorf("unexpected path, %s != %s", list[0].Path().String(), path.IpfsPath(nd3.Cid()).String())
 	}
 
-	list, err = api.Pin().Ls(ctx, opt.Pin.Type.Recursive())
+	list, err = accPins(api.Pin().Ls(ctx, opt.Pin.Type.Recursive()))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -173,10 +173,10 @@ func (tp *TestSuite) TestPinRecursive(t *testing.T) {
 	}
 
 	if list[0].Path().String() != path.IpldPath(nd2.Cid()).String() {
-		t.Errorf("unexpected path, %s != %s", list[0].Path().String(), path.IpldPath(nd3.Cid()).String())
+		t.Errorf("unexpected path, %s != %s", list[0].Path().String(), path.IpldPath(nd2.Cid()).String())
 	}
 
-	list, err = api.Pin().Ls(ctx, opt.Pin.Type.Indirect())
+	list, err = accPins(api.Pin().Ls(ctx, opt.Pin.Type.Indirect()))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -186,7 +186,7 @@ func (tp *TestSuite) TestPinRecursive(t *testing.T) {
 	}
 
 	if list[0].Path().Cid().String() != p0.Cid().String() {
-		t.Error("unexpected path")
+		t.Errorf("unexpected path, %s != %s", list[0].Path().Cid().String(), p0.Cid().String())
 	}
 
 	res, err := api.Pin().Verify(ctx)
@@ -390,21 +390,21 @@ func getThreeChainedNodes(t *testing.T, ctx context.Context, api iface.CoreAPI,
 func assertPinTypes(t *testing.T, ctx context.Context, api iface.CoreAPI, recusive, direct, indirect []cidContainer) {
 	assertPinLsAllConsistency(t, ctx, api)
 
-	list, err := api.Pin().Ls(ctx, opt.Pin.Type.Recursive())
+	list, err := accPins(api.Pin().Ls(ctx, opt.Pin.Type.Recursive()))
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	assertPinCids(t, list, recusive...)
 
-	list, err = api.Pin().Ls(ctx, opt.Pin.Type.Direct())
+	list, err = accPins(api.Pin().Ls(ctx, opt.Pin.Type.Direct()))
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	assertPinCids(t, list, direct...)
 
-	list, err = api.Pin().Ls(ctx, opt.Pin.Type.Indirect())
+	list, err = accPins(api.Pin().Ls(ctx, opt.Pin.Type.Indirect()))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -454,7 +454,7 @@ func assertPinCids(t *testing.T, pins []iface.Pin, cids ...cidContainer) {
 // assertPinLsAllConsistency verifies that listing all pins gives the same result as listing the pin types individually
 func assertPinLsAllConsistency(t *testing.T, ctx context.Context, api iface.CoreAPI) {
 	t.Helper()
-	allPins, err := api.Pin().Ls(ctx)
+	allPins, err := accPins(api.Pin().Ls(ctx))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -485,7 +485,7 @@ func assertPinLsAllConsistency(t *testing.T, ctx context.Context, api iface.Core
 	}
 
 	for typeStr, pinProps := range typeMap {
-		pins, err := api.Pin().Ls(ctx, pinProps.PinLsOption)
+		pins, err := accPins(api.Pin().Ls(ctx, pinProps.PinLsOption))
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -505,3 +505,20 @@ func assertPinLsAllConsistency(t *testing.T, ctx context.Context, api iface.Core
 		}
 	}
 }
+
+func accPins(pins <-chan iface.Pin, err error) ([]iface.Pin, error) {
+	if err != nil {
+		return nil, err
+	}
+
+	var result []iface.Pin
+
+	for pin := range pins {
+		if pin.Err() != nil {
+			return nil, pin.Err()
+		}
+		result = append(result, pin)
+	}
+
+	return result, nil
+}

tests/unixfs.go

@@ -542,7 +542,7 @@ func (tp *TestSuite) TestAddPinned(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	pins, err := api.Pin().Ls(ctx)
+	pins, err := accPins(api.Pin().Ls(ctx))
 	if err != nil {
 		t.Fatal(err)
 	}