constructor: break down libp2p logic

License: MIT
Signed-off-by: Łukasz Magiera <[email protected]>

Author: magik6k

core/node/groups.go

@@ -38,7 +38,6 @@ var BaseLibP2P = fx.Options(
 )
 
 func LibP2P(bcfg *BuildCfg, cfg *config.Config) fx.Option {
-
 	// parse ConnMgr config
 
 	grace := config.DefaultConnMgrGracePeriod

core/node/libp2p/addrs.go

@@ -0,0 +1,117 @@
+package libp2p
+
+import (
+	"fmt"
+
+	"github.com/libp2p/go-libp2p"
+	host "github.com/libp2p/go-libp2p-host"
+	p2pbhost "github.com/libp2p/go-libp2p/p2p/host/basic"
+	mafilter "github.com/libp2p/go-maddr-filter"
+	ma "github.com/multiformats/go-multiaddr"
+	mamask "github.com/whyrusleeping/multiaddr-filter"
+)
+
+func AddrFilters(filters []string) func() (opts Libp2pOpts, err error) {
+	return func() (opts Libp2pOpts, err error) {
+		for _, s := range filters {
+			f, err := mamask.NewMask(s)
+			if err != nil {
+				return opts, fmt.Errorf("incorrectly formatted address filter in config: %s", s)
+			}
+			opts.Opts = append(opts.Opts, libp2p.FilterAddresses(f))
+		}
+		return opts, nil
+	}
+}
+
+func makeAddrsFactory(announce []string, noAnnounce []string) (p2pbhost.AddrsFactory, error) {
+	var annAddrs []ma.Multiaddr
+	for _, addr := range announce {
+		maddr, err := ma.NewMultiaddr(addr)
+		if err != nil {
+			return nil, err
+		}
+		annAddrs = append(annAddrs, maddr)
+	}
+
+	filters := mafilter.NewFilters()
+	noAnnAddrs := map[string]bool{}
+	for _, addr := range noAnnounce {
+		f, err := mamask.NewMask(addr)
+		if err == nil {
+			filters.AddDialFilter(f)
+			continue
+		}
+		maddr, err := ma.NewMultiaddr(addr)
+		if err != nil {
+			return nil, err
+		}
+		noAnnAddrs[string(maddr.Bytes())] = true
+	}
+
+	return func(allAddrs []ma.Multiaddr) []ma.Multiaddr {
+		var addrs []ma.Multiaddr
+		if len(annAddrs) > 0 {
+			addrs = annAddrs
+		} else {
+			addrs = allAddrs
+		}
+
+		var out []ma.Multiaddr
+		for _, maddr := range addrs {
+			// check for exact matches
+			ok := noAnnAddrs[string(maddr.Bytes())]
+			// check for /ipcidr matches
+			if !ok && !filters.AddrBlocked(maddr) {
+				out = append(out, maddr)
+			}
+		}
+		return out
+	}, nil
+}
+
+func AddrsFactory(announce []string, noAnnounce []string) func() (opts Libp2pOpts, err error) {
+	return func() (opts Libp2pOpts, err error) {
+		addrsFactory, err := makeAddrsFactory(announce, noAnnounce)
+		if err != nil {
+			return opts, err
+		}
+		opts.Opts = append(opts.Opts, libp2p.AddrsFactory(addrsFactory))
+		return
+	}
+}
+
+func listenAddresses(addresses []string) ([]ma.Multiaddr, error) {
+	var listen []ma.Multiaddr
+	for _, addr := range addresses {
+		maddr, err := ma.NewMultiaddr(addr)
+		if err != nil {
+			return nil, fmt.Errorf("failure to parse config.Addresses.Swarm: %s", addresses)
+		}
+		listen = append(listen, maddr)
+	}
+
+	return listen, nil
+}
+
+func StartListening(addresses []string) func(host host.Host) error {
+	return func(host host.Host) error {
+		listenAddrs, err := listenAddresses(addresses)
+		if err != nil {
+			return err
+		}
+
+		// Actually start listening:
+		if err := host.Network().Listen(listenAddrs...); err != nil {
+			return err
+		}
+
+		// list out our addresses
+		addrs, err := host.Network().InterfaceListenAddresses()
+		if err != nil {
+			return err
+		}
+		log.Infof("Swarm listening at: %s", addrs)
+		return nil
+	}
+}

core/node/libp2p/host.go

@@ -0,0 +1,76 @@
+package libp2p
+
+import (
+	"context"
+
+	"github.com/libp2p/go-libp2p"
+	host "github.com/libp2p/go-libp2p-host"
+	peer "github.com/libp2p/go-libp2p-peer"
+	peerstore "github.com/libp2p/go-libp2p-peerstore"
+	record "github.com/libp2p/go-libp2p-record"
+	routing "github.com/libp2p/go-libp2p-routing"
+	routedhost "github.com/libp2p/go-libp2p/p2p/host/routed"
+	"go.uber.org/fx"
+
+	"github.com/ipfs/go-ipfs/core/node/helpers"
+	"github.com/ipfs/go-ipfs/repo"
+)
+
+type P2PHostIn struct {
+	fx.In
+
+	Repo          repo.Repo
+	Validator     record.Validator
+	HostOption    HostOption
+	RoutingOption RoutingOption
+	ID            peer.ID
+	Peerstore     peerstore.Peerstore
+
+	Opts [][]libp2p.Option `group:"libp2p"`
+}
+
+type P2PHostOut struct {
+	fx.Out
+
+	Host    host.Host
+	Routing BaseIpfsRouting
+}
+
+func Host(mctx helpers.MetricsCtx, lc fx.Lifecycle, params P2PHostIn) (out P2PHostOut, err error) {
+	opts := []libp2p.Option{libp2p.NoListenAddrs}
+	for _, o := range params.Opts {
+		opts = append(opts, o...)
+	}
+
+	ctx := helpers.LifecycleCtx(mctx, lc)
+
+	opts = append(opts, libp2p.Routing(func(h host.Host) (routing.PeerRouting, error) {
+		r, err := params.RoutingOption(ctx, h, params.Repo.Datastore(), params.Validator)
+		out.Routing = r
+		return r, err
+	}))
+
+	out.Host, err = params.HostOption(ctx, params.ID, params.Peerstore, opts...)
+	if err != nil {
+		return P2PHostOut{}, err
+	}
+
+	// this code is necessary just for tests: mock network constructions
+	// ignore the libp2p constructor options that actually construct the routing!
+	if out.Routing == nil {
+		r, err := params.RoutingOption(ctx, out.Host, params.Repo.Datastore(), params.Validator)
+		if err != nil {
+			return P2PHostOut{}, err
+		}
+		out.Routing = r
+		out.Host = routedhost.Wrap(out.Host, out.Routing)
+	}
+
+	lc.Append(fx.Hook{
+		OnStop: func(ctx context.Context) error {
+			return out.Host.Close()
+		},
+	})
+
+	return out, err
+}

core/node/libp2p/hostopt.go

@@ -0,0 +1,25 @@
+package libp2p
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/libp2p/go-libp2p"
+	host "github.com/libp2p/go-libp2p-host"
+	peer "github.com/libp2p/go-libp2p-peer"
+	peerstore "github.com/libp2p/go-libp2p-peerstore"
+)
+
+type HostOption func(ctx context.Context, id peer.ID, ps peerstore.Peerstore, options ...libp2p.Option) (host.Host, error)
+
+var DefaultHostOption HostOption = constructPeerHost
+
+// isolates the complex initialization steps
+func constructPeerHost(ctx context.Context, id peer.ID, ps peerstore.Peerstore, options ...libp2p.Option) (host.Host, error) {
+	pkey := ps.PrivKey(id)
+	if pkey == nil {
+		return nil, fmt.Errorf("missing private key for node ID: %s", id.Pretty())
+	}
+	options = append([]libp2p.Option{libp2p.Identity(pkey), libp2p.Peerstore(ps)}, options...)
+	return libp2p.New(ctx, options...)
+}