feat(dot/telemetry): implement telemetry message network_state

dot/network/service.go

@@ -19,7 +19,6 @@ package network
 import (
 	"context"
 	"errors"
-	"fmt"
 	"io"
 	"math/big"
 	"os"
@@ -305,12 +304,6 @@ func (s *Service) logPeerCount() {
 	}
 }
 
-type peerInfo struct {
-	Roles      byte   `json:"roles"`
-	BestHash   string `json:"bestHash"`
-	BestNumber uint64 `json:"bestNumber"`
-}
-
 func (s *Service) publishNetworkTelemetry(done chan interface{}) {
 	ticker := time.NewTicker(s.telemetryInterval)
 	defer ticker.Stop()
@@ -324,35 +317,11 @@ main:
 		case <-ticker.C:
 			o := s.host.bwc.GetBandwidthTotals()
 			err := telemetry.GetInstance().SendMessage(telemetry.NewBandwidthTM(o.RateIn, o.RateOut, s.host.peerCount()))
-
 			if err != nil {
 				logger.Debug("problem sending system.interval telemetry message", "error", err)
 			}
-			netState := make(map[string]interface{})
-			netState["peerId"] = s.host.h.ID()
-			hostAddrs := []string{}
-			for _, v := range s.host.h.Addrs() {
-				hostAddrs = append(hostAddrs, v.String())
-			}
-			netState["externalAddressess"] = hostAddrs
-			listAddrs := []string{}
-			for _, v := range s.host.h.Network().ListenAddresses() {
-				listAddrs = append(listAddrs, fmt.Sprintf("%s/p2p/%s", v, s.host.h.ID()))
-			}
-			netState["listenedAddressess"] = listAddrs
-
-			peers := make(map[string]interface{})
-			for _, v := range s.Peers() {
-				p := &peerInfo{
-					Roles:      v.Roles,
-					BestHash:   v.BestHash.String(),
-					BestNumber: v.BestNumber,
-				}
-				peers[v.PeerID] = *p
-			}
-			netState["connectedPeers"] = peers
 
-			err = telemetry.GetInstance().SendMessage(telemetry.NewNetworkStateTM(netState))
+			err = telemetry.GetInstance().SendMessage(telemetry.NewNetworkStateTM(s.host.h, s.Peers()))
 			if err != nil {
 				logger.Debug("problem sending system.interval telemetry message", "error", err)
 			}

dot/network/state.go

@@ -55,6 +55,6 @@ type Syncer interface {
 
 // TransactionHandler is the interface used by the transactions sub-protocol
 type TransactionHandler interface {
-	HandleTransactionMessage(*TransactionMessage) error
+	HandleTransactionMessage(*TransactionMessage) (bool, error)
 	TransactionsCount() int
 }

dot/network/transaction_test.go

@@ -55,7 +55,7 @@ func TestDecodeTransactionMessage(t *testing.T) {
 func TestHandleTransactionMessage(t *testing.T) {
 	basePath := utils.NewTestBasePath(t, "nodeA")
 	mockhandler := &MockTransactionHandler{}
-	mockhandler.On("HandleTransactionMessage", mock.AnythingOfType("*network.TransactionMessage")).Return(nil)
+	mockhandler.On("HandleTransactionMessage", mock.AnythingOfType("*network.TransactionMessage")).Return(true, nil)
 	mockhandler.On("TransactionsCount").Return(0)
 
 	config := &Config{

dot/telemetry/telemetry.go

@@ -19,6 +19,7 @@ package telemetry
 import (
 	"encoding/json"
 	"errors"
+	"fmt"
 	"math/big"
 	"sync"
 	"time"
@@ -27,6 +28,7 @@ import (
 	"github.com/ChainSafe/gossamer/lib/genesis"
 	log "github.com/ChainSafe/log15"
 	"github.com/gorilla/websocket"
+	libp2phost "github.com/libp2p/go-libp2p-core/host"
 )
 
 type telemetryConnection struct {
@@ -95,7 +97,7 @@ func (h *Handler) startListening() {
 		msg := <-h.msg
 		go func() {
 			msgBytes, err := h.msgToJSON(msg)
-			if err != nil || len(msgBytes) == 0 {
+			if err != nil {
 				h.log.Debug("issue decoding telemetry message", "error", err)
 				return
 			}
@@ -236,17 +238,47 @@ func (tm *SystemIntervalTM) messageType() string {
 	return tm.Msg
 }
 
+type peerInfo struct {
+	Roles      byte   `json:"roles"`
+	BestHash   string `json:"bestHash"`
+	BestNumber uint64 `json:"bestNumber"`
+}
+
 // NetworkStateTM struct to hold network state telemetry messages
 type NetworkStateTM struct {
 	Msg   string                 `json:"msg"`
 	State map[string]interface{} `json:"state"`
 }
 
 // NewNetworkStateTM function to create new Network State Telemetry Message
-func NewNetworkStateTM(state map[string]interface{}) *NetworkStateTM {
+func NewNetworkStateTM(host libp2phost.Host, peerInfos []common.PeerInfo) *NetworkStateTM {
+	netState := make(map[string]interface{})
+	netState["peerId"] = host.ID()
+	hostAddrs := []string{}
+	for _, v := range host.Addrs() {
+		hostAddrs = append(hostAddrs, v.String())
+	}
+	netState["externalAddressess"] = hostAddrs
+	listAddrs := []string{}
+	for _, v := range host.Network().ListenAddresses() {
+		listAddrs = append(listAddrs, fmt.Sprintf("%s/p2p/%s", v, host.ID()))
+	}
+	netState["listenedAddressess"] = listAddrs
+
+	peers := make(map[string]interface{})
+	for _, v := range peerInfos {
+		p := &peerInfo{
+			Roles:      v.Roles,
+			BestHash:   v.BestHash.String(),
+			BestNumber: v.BestNumber,
+		}
+		peers[v.PeerID] = *p
+	}
+	netState["connectedPeers"] = peers
+
 	return &NetworkStateTM{
 		Msg:   "system.network_state",
-		State: state,
+		State: netState,
 	}
 }
 func (tm *NetworkStateTM) messageType() string {