test(graphsync): test server-side graphsync

test/bin/Rules.mk

@@ -18,6 +18,11 @@ $(d)/go-sleep: github.com/ipfs/go-ipfs/test/dependencies/go-sleep
 	$(go-build-testdep)
 TGTS_$(d) += $(d)/go-sleep
 
+.PHONY: github.com/ipfs/go-ipfs/test/dependencies/graphsync-get
+$(d)/graphsync-get: github.com/ipfs/go-ipfs/test/dependencies/graphsync-get
+	$(go-build-testdep)
+TGTS_$(d) += $(d)/graphsync-get
+
 .PHONY: github.com/ipfs/go-ipfs/test/dependencies/go-timeout
 $(d)/go-timeout: github.com/ipfs/go-ipfs/test/dependencies/go-timeout
 	$(go-build-testdep)

test/dependencies/go.mod

@@ -5,17 +5,26 @@ go 1.13
 require (
 	github.com/Kubuxu/gocovmerge v0.0.0-20161216165753-7ecaa51963cd
 	github.com/golangci/golangci-lint v1.18.0
+	github.com/ipfs/go-blockservice v0.1.2
+	github.com/ipfs/go-cid v0.0.5
 	github.com/ipfs/go-cidutil v0.0.2
-	github.com/ipfs/go-log v0.0.1
+	github.com/ipfs/go-datastore v0.4.4
+	github.com/ipfs/go-graphsync v0.0.4
+	github.com/ipfs/go-ipfs-blockstore v1.0.0
+	github.com/ipfs/go-ipfs-exchange-offline v0.0.1
+	github.com/ipfs/go-log v1.0.2
+	github.com/ipfs/go-merkledag v0.3.1
+	github.com/ipfs/go-unixfs v0.2.4
 	github.com/ipfs/hang-fds v0.0.1
 	github.com/ipfs/iptb v1.4.0
-	github.com/ipfs/iptb-plugins v0.2.0
+	github.com/ipfs/iptb-plugins v0.2.1
+	github.com/ipld/go-ipld-prime v0.0.2-0.20191108012745-28a82f04c785
 	github.com/jbenet/go-random v0.0.0-20190219211222-123a90aedc0c
 	github.com/jbenet/go-random-files v0.0.0-20190219210431-31b3f20ebded
-	github.com/multiformats/go-multiaddr v0.0.4
-	github.com/multiformats/go-multiaddr-net v0.0.1
-	github.com/multiformats/go-multihash v0.0.7
-	github.com/ultraware/funlen v0.0.2 // indirect
-	golang.org/x/tools v0.0.0-20190912185636-87d9f09c5d89 // indirect
+	github.com/libp2p/go-libp2p v0.5.2
+	github.com/libp2p/go-libp2p-core v0.3.1
+	github.com/multiformats/go-multiaddr v0.2.0
+	github.com/multiformats/go-multiaddr-net v0.1.2
+	github.com/multiformats/go-multihash v0.0.13
 	gotest.tools/gotestsum v0.3.5
 )

test/dependencies/graphsync-get/graphsync-get.go

+package main
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"log"
+	"os"
+
+	"github.com/ipfs/go-blockservice"
+	"github.com/ipfs/go-cid"
+	"github.com/ipfs/go-datastore"
+	dssync "github.com/ipfs/go-datastore/sync"
+	"github.com/ipfs/go-graphsync"
+	gsimpl "github.com/ipfs/go-graphsync/impl"
+	"github.com/ipfs/go-graphsync/ipldbridge"
+	"github.com/ipfs/go-graphsync/network"
+	"github.com/ipfs/go-graphsync/storeutil"
+	"github.com/ipfs/go-ipfs-blockstore"
+	"github.com/ipfs/go-ipfs-exchange-offline"
+	"github.com/ipfs/go-merkledag"
+	uio "github.com/ipfs/go-unixfs/io"
+	"github.com/ipld/go-ipld-prime"
+	ipldfree "github.com/ipld/go-ipld-prime/impl/free"
+	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
+	ipldselector "github.com/ipld/go-ipld-prime/traversal/selector"
+	"github.com/ipld/go-ipld-prime/traversal/selector/builder"
+	"github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p-core/host"
+	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/multiformats/go-multiaddr"
+)
+
+func newGraphsync(ctx context.Context, p2p host.Host, bs blockstore.Blockstore) (graphsync.GraphExchange, error) {
+	network := network.NewFromLibp2pHost(p2p)
+	ipldBridge := ipldbridge.NewIPLDBridge()
+	return gsimpl.New(ctx,
+		network, ipldBridge,
+		storeutil.LoaderForBlockstore(bs),
+		storeutil.StorerForBlockstore(bs),
+	), nil
+}
+
+var selectAll ipld.Node = func() ipld.Node {
+	ssb := builder.NewSelectorSpecBuilder(ipldfree.NodeBuilder())
+	return ssb.ExploreRecursive(
+		ipldselector.RecursionLimitDepth(100), // default max
+		ssb.ExploreAll(ssb.ExploreRecursiveEdge()),
+	).Node()
+}()
+
+func fetch(ctx context.Context, gs graphsync.GraphExchange, p peer.ID, c cid.Cid) error {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	resps, errs := gs.Request(ctx, p, cidlink.Link{Cid: c}, selectAll)
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case _, ok := <-resps:
+			if !ok {
+				resps = nil
+			}
+		case err, ok := <-errs:
+			if !ok {
+				// done.
+				return nil
+			}
+			if err != nil {
+				return fmt.Errorf("got an unexpected error: %s", err)
+			}
+		}
+	}
+}
+
+func main() {
+	if len(os.Args) != 3 {
+		log.Fatalf("expected a multiaddr and a CID, got %d args", len(os.Args)-1)
+	}
+	addr, err := multiaddr.NewMultiaddr(os.Args[1])
+	if err != nil {
+		log.Fatalf("failed to multiaddr '%q': %s", os.Args[1], err)
+	}
+	ai, err := peer.AddrInfoFromP2pAddr(addr)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	target, err := cid.Decode(os.Args[2])
+	if err != nil {
+		log.Fatalf("failed to decode CID '%q': %s", os.Args[2], err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	p2p, err := libp2p.New(ctx, libp2p.NoListenAddrs)
+	if err != nil {
+		log.Fatal(err)
+	}
+	err = p2p.Connect(ctx, *ai)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	bs := blockstore.NewBlockstore(dssync.MutexWrap(datastore.NewMapDatastore()))
+	gs, err := newGraphsync(ctx, p2p, bs)
+	if err != nil {
+		log.Fatal("failed to start", err)
+	}
+	err = fetch(ctx, gs, ai.ID, target)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	dag := merkledag.NewDAGService(blockservice.New(bs, offline.Exchange(bs)))
+	root, err := dag.Get(ctx, target)
+	if err != nil {
+		log.Fatal(err)
+	}
+	reader, err := uio.NewDagReader(ctx, root, dag)
+	if err != nil {
+		log.Fatal(err)
+	}
+	_, err = io.Copy(os.Stdout, reader)
+	if err != nil {
+		log.Fatal(err)
+	}
+}

test/sharness/Rules.mk

@@ -7,7 +7,7 @@ T_$(d) = $(sort $(wildcard $(d)/t[0-9][0-9][0-9][0-9]-*.sh))
 DEPS_$(d) := test/bin/random test/bin/multihash test/bin/pollEndpoint \
 	   test/bin/iptb test/bin/go-sleep test/bin/random-files \
 	   test/bin/go-timeout test/bin/hang-fds test/bin/ma-pipe-unidir \
-	   test/bin/cid-fmt
+	   test/bin/cid-fmt test/bin/graphsync-get
 DEPS_$(d) += cmd/ipfs/ipfs
 DEPS_$(d) += $(d)/clean-test-results
 DEPS_$(d) += $(SHARNESS_$(d))

test/sharness/t0221-graphsync.sh

+#!/usr/bin/env bash
+
+test_description="Test fetching from graphsync."
+
+# imports
+. lib/test-lib.sh
+
+test_init_ipfs
+
+test_expect_success 'configuring ipfs' '
+  ipfs config --json Experimental.GraphsyncEnabled true
+'
+
+test_expect_success 'add content' '
+  HASH=$(random 1000000 | ipfs add -q)
+'
+
+test_launch_ipfs_daemon
+
+test_expect_success 'get addrs' '
+  ADDR="$(ipfs id --format="<addrs>" | head -1)"
+'
+
+test_expect_success 'fetch' '
+  graphsync-get "$ADDR" "$HASH" > result
+'
+
+test_expect_success 'check' '
+  ipfs add -q < result > hash_actual &&
+  echo "$HASH" > hash_expected &&
+  test_cmp hash_expected hash_actual
+'
+
+test_kill_ipfs_daemon
+test_done