Cleanup routing related units
License: MIT Signed-off-by: Łukasz Magiera <magik6k@gmail.com>
Łukasz Magiera committed
Apr 15, 2019 at 17:06 UTC
bf380b873811e06a7e54c91dc541f7e122bff2dc
2 files changed
+79
-61
core/node/groups.go
+6
-2
@@ -35,15 +35,19 @@ var BaseLibP2P = fx.Options(
35
)
36
37
func LibP2P(cfg *BuildCfg) fx.Option {
38
- return fx.Options(
38
+ opts := fx.Options(
39
BaseLibP2P,
40
41
maybeProvide(P2PNoSecurity, cfg.DisableEncryptedConnections),
42
maybeProvide(Pubsub, cfg.getOpt("pubsub") || cfg.getOpt("ipnsps")),
43
44
fx.Provide(P2PSmuxTransport(cfg.getOpt("mplex"))),
45
- fx.Provide(P2POnlineRouting(cfg.getOpt("ipnsps"))),
45
+ fx.Provide(P2PRouting),
46
+ fx.Provide(P2PBaseRouting),
47
+ maybeProvide(P2PPubsubRouter, cfg.getOpt("ipnsps")),
48
)
49
+
50
+ return opts
51
}
52
53
func Storage(cfg *BuildCfg) fx.Option {
core/node/libp2p.go
+73
-59
@@ -6,6 +6,7 @@ import (
6
"fmt"
7
"io/ioutil"
8
"os"
9
+ "sort"
10
"strings"
11
"time"
12
@@ -398,10 +399,8 @@ type P2PHostOut struct {
399
400
Host host.Host
401
Routing BaseRouting
401
- IpfsDHT *dht.IpfsDHT
402
}
403
404
-// TODO: move some of this into params struct
404
func P2PHost(mctx MetricsCtx, lc fx.Lifecycle, params P2PHostIn) (out P2PHostOut, err error) {
405
opts := []libp2p.Option{libp2p.NoListenAddrs}
406
for _, o := range params.Opts {
@@ -438,80 +437,95 @@ func P2PHost(mctx MetricsCtx, lc fx.Lifecycle, params P2PHostIn) (out P2PHostOut
437
},
438
})
439
441
- // TODO: break this up into more DI units
442
- // TODO: I'm not a fan of type assertions like this but the
443
- // `RoutingOption` system doesn't currently provide access to the
444
- // IpfsNode.
445
- //
446
- // Ideally, we'd do something like:
447
- //
448
- // 1. Add some fancy method to introspect into tiered routers to extract
449
- // things like the pubsub router or the DHT (complicated, messy,
450
- // probably not worth it).
451
- // 2. Pass the IpfsNode into the RoutingOption (would also remove the
452
- // PSRouter case below.
453
- // 3. Introduce some kind of service manager? (my personal favorite but
454
- // that requires a fair amount of work).
455
- if dht, ok := out.Routing.(*dht.IpfsDHT); ok {
456
- out.IpfsDHT = dht
440
+ return out, err
441
+}
442
+
443
+type Router struct {
444
+ routing.IpfsRouting
445
+
446
+ Priority int // less = more important
447
+}
448
+
449
+type p2pRouterOut struct {
450
+ fx.Out
451
+
452
+ Router Router `group:"routers"`
453
+}
454
+
455
+func P2PBaseRouting(lc fx.Lifecycle, in BaseRouting) (out p2pRouterOut, dr *dht.IpfsDHT) {
456
+ if dht, ok := in.(*dht.IpfsDHT); ok {
457
+ dr = dht
458
459
lc.Append(fx.Hook{
460
OnStop: func(ctx context.Context) error {
460
- return out.IpfsDHT.Close()
461
+ return dr.Close()
462
},
463
})
464
}
465
465
- return out, err
466
+ return p2pRouterOut{
467
+ Router: Router{
468
+ Priority: 1000,
469
+ IpfsRouting: in,
470
+ },
471
+ }, dr
472
}
473
468
-type p2pRoutingIn struct {
474
+type p2pOnlineRoutingIn struct {
475
fx.In
476
471
- Repo repo.Repo
477
+ Routers []Router `group:"routers"`
478
Validator record.Validator
473
- Host host.Host
474
- PubSub *pubsub.PubSub `optional:"true"`
479
+}
480
476
- BaseRouting BaseRouting
481
+func P2PRouting(in p2pOnlineRoutingIn) routing.IpfsRouting {
482
+ routers := in.Routers
483
+
484
+ sort.SliceStable(routers, func(i, j int) bool {
485
+ return routers[i].Priority < routers[j].Priority
486
+ })
487
+
488
+ irouters := make([]routing.IpfsRouting, len(routers))
489
+ for i, v := range routers {
490
+ irouters[i] = v.IpfsRouting
491
+ }
492
+
493
+ return routinghelpers.Tiered{
494
+ Routers: irouters,
495
+ Validator: in.Validator,
496
+ }
497
}
498
479
-type p2pRoutingOut struct {
480
- fx.Out
499
+type p2pPSRoutingIn struct {
500
+ fx.In
501
482
- IpfsRouting routing.IpfsRouting
483
- PSRouter *namesys.PubsubValueStore
484
-}
485
-
486
-func P2POnlineRouting(ipnsps bool) func(mctx MetricsCtx, lc fx.Lifecycle, in p2pRoutingIn) (out p2pRoutingOut) {
487
- return func(mctx MetricsCtx, lc fx.Lifecycle, in p2pRoutingIn) (out p2pRoutingOut) {
488
- out.IpfsRouting = in.BaseRouting
489
-
490
- if ipnsps {
491
- out.PSRouter = namesys.NewPubsubValueStore(
492
- lifecycleCtx(mctx, lc),
493
- in.Host,
494
- in.BaseRouting,
495
- in.PubSub,
496
- in.Validator,
497
- )
498
-
499
- out.IpfsRouting = routinghelpers.Tiered{
500
- Routers: []routing.IpfsRouting{
501
- // Always check pubsub first.
502
- &routinghelpers.Compose{
503
- ValueStore: &routinghelpers.LimitedValueStore{
504
- ValueStore: out.PSRouter,
505
- Namespaces: []string{"ipns"},
506
- },
507
- },
508
- in.BaseRouting,
502
+ BaseRouting BaseRouting
503
+ Repo repo.Repo
504
+ Validator record.Validator
505
+ Host host.Host
506
+ PubSub *pubsub.PubSub `optional:"true"`
507
+}
508
+
509
+func P2PPubsubRouter(mctx MetricsCtx, lc fx.Lifecycle, in p2pPSRoutingIn) (p2pRouterOut, *namesys.PubsubValueStore) {
510
+ psRouter := namesys.NewPubsubValueStore(
511
+ lifecycleCtx(mctx, lc),
512
+ in.Host,
513
+ in.BaseRouting,
514
+ in.PubSub,
515
+ in.Validator,
516
+ )
517
+
518
+ return p2pRouterOut{
519
+ Router: Router{
520
+ IpfsRouting: &routinghelpers.Compose{
521
+ ValueStore: &routinghelpers.LimitedValueStore{
522
+ ValueStore: psRouter,
523
+ Namespaces: []string{"ipns"},
524
},
510
- Validator: in.Validator,
511
- }
512
- }
513
- return out
514
- }
525
+ },
526
+ Priority: 100,
527
+ },
528
+ }, psRouter
529
}
530
531
func AutoNATService(mctx MetricsCtx, lc fx.Lifecycle, cfg *config.Config, host host.Host) error {