Add pubsub discovery.

Author: nasdf

go.mod

@@ -25,6 +25,7 @@ require (
 	github.com/libp2p/go-libp2p-blankhost v0.2.0
 	github.com/libp2p/go-libp2p-connmgr v0.2.4
 	github.com/libp2p/go-libp2p-core v0.7.0
+	github.com/libp2p/go-libp2p-discovery v0.5.0
 	github.com/libp2p/go-libp2p-kad-dht v0.11.0
 	github.com/libp2p/go-libp2p-noise v0.1.1
 	github.com/libp2p/go-libp2p-pubsub v0.4.1

p2p/host.go

@@ -2,13 +2,16 @@ package p2p
 
 import (
 	"context"
+	"sync"
 	"time"
 
 	"github.com/libp2p/go-libp2p"
 	"github.com/libp2p/go-libp2p-connmgr"
 	"github.com/libp2p/go-libp2p-core/crypto"
 	"github.com/libp2p/go-libp2p-core/host"
+	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/routing"
+	"github.com/libp2p/go-libp2p-kad-dht"
 	"github.com/libp2p/go-libp2p-kad-dht/dual"
 	"github.com/libp2p/go-libp2p-noise"
 	"github.com/libp2p/go-libp2p-tls"
@@ -30,8 +33,8 @@ const (
 
 // NewHost returns a new libp2p host and router.
 func NewHost(ctx context.Context, priv crypto.PrivKey) (host.Host, routing.Routing, error) {
-	var err error
 	var router routing.Routing
+	var err error
 
 	host, err := libp2p.New(ctx,
 		libp2p.Identity(priv),
@@ -55,3 +58,16 @@ func NewHost(ctx context.Context, priv crypto.PrivKey) (host.Host, routing.Routi
 
 	return host, router, err
 }
+
+// Bootstrap initiates connections to a list of known peers.
+func Bootstrap(ctx context.Context, host host.Host) {
+	var wg sync.WaitGroup
+	for _, info := range dht.GetDefaultBootstrapPeerAddrInfos() {
+		wg.Add(1)
+		go func(info peer.AddrInfo) {
+			defer wg.Done()
+			host.Connect(ctx, info)
+		}(info)
+	}
+	wg.Wait()
+}

p2p/namesys.go

@@ -7,6 +7,8 @@ import (
 
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/libp2p/go-libp2p-core/routing"
+	discovery "github.com/libp2p/go-libp2p-discovery"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	namesys "github.com/libp2p/go-libp2p-pubsub-router"
 	record "github.com/libp2p/go-libp2p-record"
@@ -36,8 +38,13 @@ func PeerIDForTopic(topic string) (peer.ID, error) {
 }
 
 // NewSystem returns a new pubsub name system.
-func NewNamesys(ctx context.Context, host host.Host) (*namesys.PubsubValueStore, error) {
-	sub, err := pubsub.NewGossipSub(ctx, host)
+func NewNamesys(ctx context.Context, host host.Host, router routing.ContentRouting) (*namesys.PubsubValueStore, error) {
+	var options []pubsub.Option
+	if router != nil {
+		options = append(options, pubsub.WithDiscovery(discovery.NewRoutingDiscovery(router)))
+	}
+
+	sub, err := pubsub.NewGossipSub(ctx, host, options...)
 	if err != nil {
 		return nil, err
 	}

peer/mock.go

@@ -37,7 +37,7 @@ func NewMock(t *testing.T, ctx context.Context) *Mock {
 		t.Fatal("failed to create peer config")
 	}
 
-	namesys, err := p2p.NewNamesys(ctx, host)
+	namesys, err := p2p.NewNamesys(ctx, host, nil)
 	if err != nil {
 		t.Fatal("failed to create peer namesys")
 	}

peer/node.go

@@ -16,7 +16,6 @@ import (
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/routing"
-	"github.com/libp2p/go-libp2p-kad-dht"
 	"github.com/multiverse-vcs/go-multiverse/p2p"
 )
 
@@ -47,11 +46,6 @@ func New(ctx context.Context, dstore datastore.Batching, config *Config) (*Node,
 		return nil, err
 	}
 
-	namesys, err := p2p.NewNamesys(ctx, host)
-	if err != nil {
-		return nil, err
-	}
-
 	bstore := blockstore.NewBlockstore(dstore)
 	net := bsnet.NewFromIpfsHost(host, router)
 	exc := bitswap.New(ctx, net, bstore)
@@ -63,16 +57,23 @@ func New(ctx context.Context, dstore datastore.Batching, config *Config) (*Node,
 	if err != nil {
 		return nil, err
 	}
+
+	p2p.Bootstrap(ctx, host)
 	provsys.Run()
 
-	for _, info := range dht.GetDefaultBootstrapPeerAddrInfos() {
-		go host.Connect(ctx, info)
+	namesys, err := p2p.NewNamesys(ctx, host, router)
+	if err != nil {
+		return nil, err
 	}
 
 	if err := p2p.Discovery(ctx, host); err != nil {
 		return nil, err
 	}
 
+	if err := router.Bootstrap(ctx); err != nil {
+		return nil, err
+	}
+
 	return &Node{
 		dag:     dag,
 		host:    host,
@@ -96,6 +97,16 @@ func (n *Node) Config() *Config {
 	return n.config
 }
 
+// Connect connects to the peer with the given ID.
+func (n *Node) Connect(ctx context.Context, id peer.ID) error {
+	info, err := n.router.FindPeer(ctx, id)
+	if err != nil {
+		return nil
+	}
+
+	return n.host.Connect(ctx, info)
+}
+
 // Dag returns the merkledag api.
 func (n *Node) Dag() ipld.DAGService {
 	return n.dag

peer/peer.go

@@ -23,5 +23,5 @@ type Peer interface {
 	// Namesys returns the name system.
 	Namesys() routing.ValueStore
 	// ResolvePath resolves the node from the given path.
-	ResolvePath(ctx context.Context, p path.Path) (ipld.Node, error)
+	ResolvePath(context.Context, path.Path) (ipld.Node, error)
 }