netsync, main: add ConnectedPeers to Manager

ConnectedPeers returns all the currently connected peers and any new peer that's additionally connected through the returned channel. This method is required for query.Workmanager as it needs ot receive peers that it can request blocks from.
btcsuite · kcalvinalvin · Aug 6, 2024 · Aug 6, 2024 · Nov 29, 2024 · Aug 7, 2024
commit 0a9f2072d333a51da43a4189daf9adcc4f2e27db
diff --git a/netsync/interface.go b/netsync/interface.go
@@ -38,4 +38,7 @@ type Config struct {
 	MaxPeers           int
 
 	FeeEstimator *mempool.FeeEstimator
+
+	// ConnectedPeers returns all the currently connected peers.
+	ConnectedPeers func() []*peer.Peer
 }
diff --git a/netsync/manager.go b/netsync/manager.go
@@ -295,6 +295,7 @@ type SyncManager struct {
 	queuedBlocks         map[chainhash.Hash]*blockMsg
 	queuedBlocksPrevHash map[chainhash.Hash]chainhash.Hash
 	peerSubscribers      []*peerSubscription
+	connectedPeers       func() []*peerpkg.Peer
 
 	// An optional fee estimator.
 	feeEstimator *mempool.FeeEstimator
@@ -1919,6 +1920,29 @@ type peerSubscription struct {
 	cancel <-chan struct{}
 }
 
+// ConnectedPeers returns all the currently connected peers to the channel
+// and then any additional new peers on connect.
+func (sm *SyncManager) ConnectedPeers() (<-chan query.Peer, func(), error) {
+	peers := sm.connectedPeers()
+	peerChan := make(chan query.Peer, len(peers))
+
+	for _, peer := range peers {
+		if sm.isSyncCandidate(peer) {
+			peerChan <- peer
+		}
+	}
+
+	cancelChan := make(chan struct{})
+	sm.peerSubscribers = append(sm.peerSubscribers, &peerSubscription{
+		peers:  peerChan,
+		cancel: cancelChan,
+	})
+
+	return peerChan, func() {
+		close(cancelChan)
+	}, nil
+}
+
 // New constructs a new SyncManager. Use Start to begin processing asynchronous
 // block, tx, and inv updates.
 func New(config *Config) (*SyncManager, error) {
@@ -1938,6 +1962,7 @@ func New(config *Config) (*SyncManager, error) {
 		queuedBlocks:         make(map[chainhash.Hash]*blockMsg),
 		queuedBlocksPrevHash: make(map[chainhash.Hash]chainhash.Hash),
 		feeEstimator:         config.FeeEstimator,
+		connectedPeers:       config.ConnectedPeers,
 	}
 
 	best := sm.chain.BestSnapshot()

diff --git a/server.go b/server.go
@@ -2940,6 +2940,7 @@ func newServer(listenAddrs, agentBlacklist, agentWhitelist []string,
 		DisableCheckpoints: cfg.DisableCheckpoints,
 		MaxPeers:           cfg.MaxPeers,
 		FeeEstimator:       s.feeEstimator,
+		ConnectedPeers:     s.ConnectedPeers,
 	})
 	if err != nil {
 		return nil, err