Skip to content
Snippets Groups Projects
Select Git revision
  • 8e22cef581c585332baa4f719330061edecc334b
  • without_tipselection default
  • develop protected
  • fix/grafana-local-dashboard
  • wasp
  • fix/dashboard-explorer-freeze
  • master
  • feat/timerqueue
  • test/sync_debug_and_650
  • feat/sync_revamp_inv
  • wip/sync
  • tool/db-recovery
  • portcheck/fix
  • fix/synchronization
  • feat/new-dashboard-analysis
  • feat/refactored-analysis-dashboard
  • feat/new-analysis-dashboard
  • test/demo-prometheus-fpc
  • prometheus_metrics
  • wip/analysis-server
  • merge/fpc-test-value-transfer
  • v0.2.2
  • v0.2.1
  • v0.2.0
  • v0.1.3
  • v0.1.2
  • v0.1.1
  • v0.1.0
28 results

plugin.go

Blame
  • user avatar
    Wolfgang Welz authored and GitHub committed
    * adapt to hive.go autopeering events changes
    
    * add exposed function to start autopeering neighbor selection
    
    * start autopeering neighbor selection when gossip is started
    
    * update go modules
    
    * Apply suggestions from code review
    
    Co-Authored-By: default avatarAngelo Capossele <angelocapossele@gmail.com>
    
    * fix build errors
    
    Co-authored-by: default avatarAngelo Capossele <angelocapossele@gmail.com>
    8e22cef5
    History
    plugin.go 3.20 KiB
    package client
    
    import (
    	"encoding/hex"
    	"net"
    	"strings"
    	"sync"
    	"time"
    
    	"github.com/iotaledger/goshimmer/packages/shutdown"
    	"github.com/iotaledger/goshimmer/plugins/analysis/types/heartbeat"
    	"github.com/iotaledger/goshimmer/plugins/autopeering"
    	"github.com/iotaledger/goshimmer/plugins/autopeering/local"
    	"github.com/iotaledger/goshimmer/plugins/config"
    	"github.com/iotaledger/hive.go/daemon"
    	"github.com/iotaledger/hive.go/logger"
    	"github.com/iotaledger/hive.go/network"
    	"github.com/iotaledger/hive.go/node"
    )
    
    var log *logger.Logger
    var connLock sync.Mutex
    
    // Run runs the plugin.
    func Run(plugin *node.Plugin) {
    	log = logger.NewLogger("Analysis-Client")
    	daemon.BackgroundWorker("Analysis Client", func(shutdownSignal <-chan struct{}) {
    		ticker := time.NewTicker(ReportInterval * time.Second)
    		defer ticker.Stop()
    		for {
    			select {
    			case <-shutdownSignal:
    				return
    
    			case <-ticker.C:
    				conn, err := net.Dial("tcp", config.Node.GetString(CfgServerAddress))
    				if err != nil {
    					log.Debugf("Could not connect to reporting server: %s", err.Error())
    					continue
    				}
    				managedConn := network.NewManagedConnection(conn)
    				eventDispatchers := getEventDispatchers(managedConn)
    				reportHeartbeat(eventDispatchers)
    			}
    		}
    	}, shutdown.PriorityAnalysis)
    }
    
    func getEventDispatchers(conn *network.ManagedConnection) *EventDispatchers {
    	return &EventDispatchers{
    		Heartbeat: func(packet *heartbeat.Packet) {
    			var out strings.Builder
    			for _, value := range packet.OutboundIDs {
    				out.WriteString(hex.EncodeToString(value))
    			}
    			var in strings.Builder
    			for _, value := range packet.InboundIDs {
    				in.WriteString(hex.EncodeToString(value))
    			}
    			log.Debugw(
    				"Heartbeat",
    				"nodeID", hex.EncodeToString(packet.OwnID),
    				"outboundIDs", out.String(),
    				"inboundIDs", in.String(),
    			)
    
    			// Marshal() copies the content of packet, it doesn't modify it.
    			data, err := packet.Marshal()
    			if err != nil {
    				log.Info(err, " - heartbeat message skipped")
    				return
    			}
    
    			connLock.Lock()
    			defer connLock.Unlock()
    			if _, err = conn.Write(data); err != nil {
    				log.Debugw("Error while writing to connection", "Description", err)
    			}
    		},
    	}
    }
    
    func reportHeartbeat(dispatchers *EventDispatchers) {
    	// Get own ID
    	var nodeID []byte
    	if local.GetInstance() != nil {
    		// Doesn't copy the ID, take care not to modify underlying bytearray!
    		nodeID = local.GetInstance().ID().Bytes()
    	}
    
    	var outboundIDs [][]byte
    	var inboundIDs [][]byte
    
    	// Get outboundIDs (chosen neighbors)
    	outgoingNeighbors := autopeering.Selection().GetOutgoingNeighbors()
    	outboundIDs = make([][]byte, len(outgoingNeighbors))
    	for i, neighbor := range outgoingNeighbors {
    		// Doesn't copy the ID, take care not to modify underlying bytearray!
    		outboundIDs[i] = neighbor.ID().Bytes()
    	}
    
    	// Get inboundIDs (accepted neighbors)
    	incomingNeighbors := autopeering.Selection().GetIncomingNeighbors()
    	inboundIDs = make([][]byte, len(incomingNeighbors))
    	for i, neighbor := range incomingNeighbors {
    		// Doesn't copy the ID, take care not to modify underlying bytearray!
    		inboundIDs[i] = neighbor.ID().Bytes()
    	}
    
    	packet := &heartbeat.Packet{OwnID: nodeID, OutboundIDs: outboundIDs, InboundIDs: inboundIDs}
    
    	dispatchers.Heartbeat(packet)
    }