graphsync-get.go

package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"os"

	"gitlab.dms3.io/dms3/go-blockservice"
	"gitlab.dms3.io/dms3/go-cid"
	"gitlab.dms3.io/dms3/go-datastore"
	dssync "gitlab.dms3.io/dms3/go-datastore/sync"
	blockstore "gitlab.dms3.io/dms3/go-dms3-blockstore"
	offline "gitlab.dms3.io/dms3/go-dms3-exchange-offline"
	"gitlab.dms3.io/dms3/go-graphsync"
	gsimpl "gitlab.dms3.io/dms3/go-graphsync/impl"
	"gitlab.dms3.io/dms3/go-graphsync/network"
	"gitlab.dms3.io/dms3/go-graphsync/storeutil"
	"gitlab.dms3.io/dms3/go-merkledag"
	uio "gitlab.dms3.io/dms3/go-unixfs/io"
	"gitlab.dms3.io/dms3/ld/go-ld-prime"
	cidlink "gitlab.dms3.io/dms3/ld/go-ld-prime/linking/cid"
	basicnode "gitlab.dms3.io/dms3/ld/go-ld-prime/node/basic"
	ldselector "gitlab.dms3.io/dms3/ld/go-ld-prime/traversal/selector"
	"gitlab.dms3.io/dms3/ld/go-ld-prime/traversal/selector/builder"
	"gitlab.dms3.io/mf/go-multiaddr"
	"gitlab.dms3.io/p2p/go-p2p"
	"gitlab.dms3.io/p2p/go-p2p-core/host"
	"gitlab.dms3.io/p2p/go-p2p-core/peer"
)

func newGraphsync(ctx context.Context, p2p host.Host, bs blockstore.Blockstore) (graphsync.GraphExchange, error) {
	network := network.NewFromP2pHost(p2p)
	return gsimpl.New(ctx,
		network,
		storeutil.LoaderForBlockstore(bs),
		storeutil.StorerForBlockstore(bs),
	), nil
}

var selectAll ld.Node = func() ld.Node {
	ssb := builder.NewSelectorSpecBuilder(basicnode.Style.Any)
	return ssb.ExploreRecursive(
		ldselector.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 := p2p.New(ctx, p2p.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)
	}
}