Add signals for notifying react about discovery activity (#859)

* Add signal for notifying react on discovery activity

* Whitelist expected signals in e2e tests

geth/peers/README.md

@@ -0,0 +1,48 @@
+Peer pool signals
+=================
+
+Peer pool sends 3 types of signals.
+
+Discovery started signal will be sent once discovery server is started.
+And every time node will have to re-start discovery server because peer number dropped too low.
+
+```json
+{
+  "type": "discovery.started",
+  "event": null
+}
+```
+
+
+Discovery stopped signal will be sent once discovery found max limit of peers
+for every registered topic.
+
+```json
+{
+  "type": "discovery.stopped",
+  "event": null
+}
+```
+
+
+Discovery summary signal will be sent every time new peer is added or removed
+from a cluster. It will contain a map with capability as a key and total numbers
+of peers with that capability as a value.
+
+```json
+{
+  "type": "discovery.summary",
+  "event": {
+    "shh/6": 1
+  }
+}
+```
+
+Or if we don't have any peers:
+
+```json
+{
+  "type": "discovery.summary",
+  "event": {}
+}
+```

geth/peers/peerpool.go

@@ -32,11 +32,6 @@ const (
 	DefaultFastSync = 3 * time.Second
 	// DefaultSlowSync is a recommended value for slow (background) peers search.
 	DefaultSlowSync = 30 * time.Minute
-
-	// Discv5Closed is sent when discv5 is closed
-	Discv5Closed PoolEvent = "discv5.closed"
-	// Discv5Started is sent when discv5 is started
-	Discv5Started PoolEvent = "discv5.started"
 )
 
 // NewPeerPool creates instance of PeerPool
@@ -74,8 +69,6 @@ type PeerPool struct {
 	quit               chan struct{}
 
 	wg sync.WaitGroup
-
-	feed event.Feed
 }
 
 // Start creates topic pool for each topic in config and subscribes to server events.
@@ -94,6 +87,7 @@ func (p *PeerPool) Start(server *p2p.Server) error {
 		}
 		p.topics = append(p.topics, topicPool)
 	}
+	SendDiscoveryStarted() // discovery must be started when pool is started
 
 	events := make(chan *p2p.PeerEvent, 20)
 	p.serverSubscription = server.SubscribeEvents(events)
@@ -115,7 +109,7 @@ func (p *PeerPool) restartDiscovery(server *p2p.Server) error {
 		}
 		log.Debug("restarted discovery from peer pool")
 		server.DiscV5 = ntab
-		p.feed.Send(Discv5Started)
+		SendDiscoveryStarted()
 	}
 	for _, t := range p.topics {
 		if !t.BelowMin() || t.SearchRunning() {
@@ -152,6 +146,7 @@ func (p *PeerPool) handleServerPeers(server *p2p.Server, events <-chan *p2p.Peer
 			case p2p.PeerEventTypeAdd:
 				p.handleAddedEvent(server, event)
 			}
+			SendDiscoverySummary(server.PeersInfo())
 		}
 	}
 }
@@ -167,7 +162,7 @@ func (p *PeerPool) handleAddedEvent(server *p2p.Server, event *p2p.PeerEvent) {
 		log.Debug("closing discv5 connection", "server", server.Self())
 		server.DiscV5.Close()
 		server.DiscV5 = nil
-		p.feed.Send(Discv5Closed)
+		SendDiscoveryStopped()
 	}
 }
 

geth/peers/peerpool_test.go

@@ -1,6 +1,7 @@
 package peers
 
 import (
+	"encoding/json"
 	"fmt"
 	"net"
 	"strconv"
@@ -12,10 +13,12 @@ import (
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/discv5"
+	"github.com/ethereum/go-ethereum/whisper/whisperv6"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 
 	"github.com/status-im/status-go/geth/params"
+	"github.com/status-im/status-go/geth/signal"
 )
 
 type PeerPoolSimulationSuite struct {
@@ -52,6 +55,7 @@ func (s *PeerPoolSimulationSuite) SetupTest() {
 	s.peers = make([]*p2p.Server, 3)
 	for i := range s.peers {
 		key, _ := crypto.GenerateKey()
+		whisper := whisperv6.New(nil)
 		peer := &p2p.Server{
 			Config: p2p.Config{
 				MaxPeers:         10,
@@ -61,6 +65,7 @@ func (s *PeerPoolSimulationSuite) SetupTest() {
 				DiscoveryV5:      true,
 				NoDiscovery:      true,
 				BootstrapNodesV5: []*discv5.Node{bootnodeV5},
+				Protocols:        whisper.Protocols(),
 			},
 		}
 		port++
@@ -82,21 +87,43 @@ func (s *PeerPoolSimulationSuite) getPeerFromEvent(events <-chan *p2p.PeerEvent,
 	return
 }
 
-func (s *PeerPoolSimulationSuite) getPoolEvent(events <-chan PoolEvent) PoolEvent {
+func (s *PeerPoolSimulationSuite) getPoolEvent(events <-chan string) string {
 	select {
 	case ev := <-events:
 		return ev
-	case <-time.After(200 * time.Millisecond):
+	case <-time.After(time.Second):
 		s.Fail("timed out waiting for a peer")
 		return ""
 	}
 }
 
 func (s *PeerPoolSimulationSuite) TestSingleTopicDiscoveryWithFailover() {
+	poolEvents := make(chan string, 1)
+	summaries := make(chan map[string]int, 1)
+	signal.SetDefaultNodeNotificationHandler(func(jsonEvent string) {
+		fmt.Println(jsonEvent)
+		var envelope struct {
+			Type  string
+			Event json.RawMessage
+		}
+		s.NoError(json.Unmarshal([]byte(jsonEvent), &envelope))
+		switch envelope.Type {
+		case DiscoveryStarted:
+			poolEvents <- envelope.Type
+		case DiscoveryStopped:
+			poolEvents <- envelope.Type
+		case DiscoverySummary:
+			poolEvents <- envelope.Type
+			var summary map[string]int
+			s.NoError(json.Unmarshal(envelope.Event, &summary))
+			summaries <- summary
+		}
+
+	})
 	topic := discv5.Topic("cap=test")
 	// simulation should only rely on fast sync
 	config := map[discv5.Topic]params.Limits{
-		topic: params.NewLimits(1, 1), // limits a chosen for simplicity of the simulation
+		topic: params.NewLimits(1, 1), // limits are chosen for simplicity of the simulation
 	}
 	peerPool := NewPeerPool(config, 100*time.Millisecond, 100*time.Millisecond, nil, true)
 	register := NewRegister(topic)
@@ -109,23 +136,39 @@ func (s *PeerPoolSimulationSuite) TestSingleTopicDiscoveryWithFailover() {
 	defer subscription.Unsubscribe()
 	s.NoError(peerPool.Start(s.peers[1]))
 	defer peerPool.Stop()
-	poolEvents := make(chan PoolEvent)
+	s.Equal(DiscoveryStarted, s.getPoolEvent(poolEvents))
-	poolSub := peerPool.feed.Subscribe(poolEvents)
-	defer poolSub.Unsubscribe()
 	connected := s.getPeerFromEvent(events, p2p.PeerEventTypeAdd)
 	s.Equal(s.peers[0].Self().ID, connected)
-	s.Equal(Discv5Closed, s.getPoolEvent(poolEvents))
+	s.Equal(DiscoveryStopped, s.getPoolEvent(poolEvents))
 	s.Require().Nil(s.peers[1].DiscV5)
+
+	s.Require().Equal(DiscoverySummary, s.getPoolEvent(poolEvents))
+	summary := <-summaries
+	s.Len(summary, 1)
+	s.Contains(summary, "shh/6")
+	s.Equal(summary["shh/6"], 1)
+
 	s.peers[0].Stop()
 	disconnected := s.getPeerFromEvent(events, p2p.PeerEventTypeDrop)
 	s.Equal(connected, disconnected)
-	s.Equal(Discv5Started, s.getPoolEvent(poolEvents))
+
+	s.Require().Equal(DiscoverySummary, s.getPoolEvent(poolEvents))
+	summary = <-summaries
+	s.Len(summary, 0)
+
+	s.Equal(DiscoveryStarted, s.getPoolEvent(poolEvents))
 	s.Require().NotNil(s.peers[1].DiscV5)
 	register = NewRegister(topic)
 	s.Require().NoError(register.Start(s.peers[2]))
 	defer register.Stop()
-	newConnected := s.getPeerFromEvent(events, p2p.PeerEventTypeAdd)
+	s.Equal(s.peers[2].Self().ID, s.getPeerFromEvent(events, p2p.PeerEventTypeAdd))
-	s.Equal(s.peers[2].Self().ID, newConnected)
+
+	s.Equal(DiscoveryStopped, s.getPoolEvent(poolEvents))
+	s.Require().Equal(DiscoverySummary, s.getPoolEvent(poolEvents))
+	summary = <-summaries
+	s.Len(summary, 1)
+	s.Contains(summary, "shh/6")
+	s.Equal(summary["shh/6"], 1)
 }
 
 func (s *PeerPoolSimulationSuite) TearDown() {

geth/peers/signal.go

@@ -0,0 +1,45 @@
+package peers
+
+import (
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/status-im/status-go/geth/signal"
+)
+
+const (
+	// DiscoveryStarted is sent when node discv5 was started.
+	DiscoveryStarted = "discovery.started"
+	// DiscoveryStopped is sent when discv5 server was stopped.
+	DiscoveryStopped = "discovery.stopped"
+
+	// DiscoverySummary is sent when peer is added or removed.
+	// it will be a map with capability=peer count k/v's.
+	DiscoverySummary = "discovery.summary"
+)
+
+// SendDiscoveryStarted sends discovery.started signal.
+func SendDiscoveryStarted() {
+	signal.Send(signal.Envelope{
+		Type: DiscoveryStarted,
+	})
+}
+
+// SendDiscoveryStopped sends discovery.stopped signal.
+func SendDiscoveryStopped() {
+	signal.Send(signal.Envelope{
+		Type: DiscoveryStopped,
+	})
+}
+
+// SendDiscoverySummary sends discovery.summary signal.
+func SendDiscoverySummary(peers []*p2p.PeerInfo) {
+	summary := map[string]int{}
+	for i := range peers {
+		for _, cap := range peers[i].Caps {
+			summary[cap]++
+		}
+	}
+	signal.Send(signal.Envelope{
+		Type:  DiscoverySummary,
+		Event: summary,
+	})
+}

t/e2e/api/api_test.go

@@ -201,6 +201,14 @@ func (s *APITestSuite) TestEventsNodeStartStop() {
 		var envelope signal.Envelope
 		err := json.Unmarshal([]byte(jsonEvent), &envelope)
 		s.NoError(err)
+		// whitelist types that we are interested in
+		switch envelope.Type {
+		case signal.EventNodeStarted:
+		case signal.EventNodeStopped:
+		case signal.EventNodeReady:
+		default:
+			return
+		}
 		envelopes <- envelope
 	})