peernotifier: Add peer notifier package for peer online/offline events

This commit adds a peer notifier package which provides clients with
a subscription to peer online and offline events.
This commit is contained in:
carla committed 2019-08-02 10:15:28 -04:00
1 parent 8c9c4b52e8
commit 4ceceda757
4 files changed
+150 -2

No files matched your search

+30 -2
View File
@@ -44,6 +44,7 @@ import (
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/nat"
"github.com/lightningnetwork/lnd/netann"
"github.com/lightningnetwork/lnd/peernotifier"
"github.com/lightningnetwork/lnd/pool"
"github.com/lightningnetwork/lnd/routing"
"github.com/lightningnetwork/lnd/routing/route"
@@ -188,6 +189,8 @@ type server struct {
channelNotifier *channelnotifier.ChannelNotifier
peerNotifier *peernotifier.PeerNotifier
witnessBeacon contractcourt.WitnessBeacon
breachArbiter *breachArbiter
@@ -1081,6 +1084,10 @@ func newServer(listenAddrs []net.Addr, chanDB *channeldb.DB,
return nil, err
}
// Assemble a peer notifier which will provide clients with subscriptions
// to peer online and offline events.
s.peerNotifier = peernotifier.New()
if cfg.WtClient.Active {
policy := wtpolicy.DefaultPolicy()
@@ -1183,6 +1190,10 @@ func (s *server) Start() error {
startErr = err
return
}
if err := s.peerNotifier.Start(); err != nil {
startErr = err
return
}
if err := s.sphinx.Start(); err != nil {
startErr = err
return
@@ -1341,6 +1352,7 @@ func (s *server) Stop() error {
s.chainArb.Stop()
s.sweeper.Stop()
s.channelNotifier.Stop()
s.peerNotifier.Stop()
s.cc.wallet.Shutdown()
s.cc.chainView.Stop()
s.connMgr.Stop()
@@ -2713,7 +2725,8 @@ func (s *server) addPeer(p *peer) {
// TODO(roasbeef): pipe all requests through to the
// queryHandler/peerManager
pubStr := string(p.addr.IdentityKey.SerializeCompressed())
pubSer := p.addr.IdentityKey.SerializeCompressed()
pubStr := string(pubSer)
s.peersByPub[pubStr] = p
@@ -2722,6 +2735,13 @@ func (s *server) addPeer(p *peer) {
} else {
s.outboundPeers[pubStr] = p
}
// Inform the peer notifier of a peer online event so that it can be reported
// to clients listening for peer events.
var pubKey [33]byte
copy(pubKey[:], pubSer)
s.peerNotifier.NotifyPeerOnline(pubKey)
}
// peerInitializer asynchronously starts a newly connected peer after it has
@@ -2963,7 +2983,8 @@ func (s *server) removePeer(p *peer) {
return
}
pubStr := string(p.addr.IdentityKey.SerializeCompressed())
pubSer := p.addr.IdentityKey.SerializeCompressed()
pubStr := string(pubSer)
delete(s.peersByPub, pubStr)
@@ -2972,6 +2993,13 @@ func (s *server) removePeer(p *peer) {
} else {
delete(s.outboundPeers, pubStr)
}
// Inform the peer notifier of a peer offline event so that it can be
// reported to clients listening for peer events.
var pubKey [33]byte
copy(pubKey[:], pubSer)
s.peerNotifier.NotifyPeerOffline(pubKey)
}
// openChanReq is a message sent to the server in order to request the