woop/api/service/networkinfo/service.go

package networkinfo

import (
	"context"
	"fmt"
	"net"
	"sync"
	"time"

	msg_pb "github.com/harmony-one/harmony/api/proto/message"
	"github.com/harmony-one/harmony/internal/utils"
	"github.com/harmony-one/harmony/p2p"
	libp2pdis "github.com/libp2p/go-libp2p-discovery"
	libp2pdht "github.com/libp2p/go-libp2p-kad-dht"
	peerstore "github.com/libp2p/go-libp2p-peerstore"
	manet "github.com/multiformats/go-multiaddr-net"
)

// Service is the network info service.
type Service struct {
	Host        p2p.Host
	Rendezvous  p2p.GroupID
	bootnodes   utils.AddrList
	dht         *libp2pdht.IpfsDHT
	ctx         context.Context
	cancel      context.CancelFunc
	stopChan    chan struct{}
	stoppedChan chan struct{}
	peerChan    chan p2p.Peer
	peerInfo    <-chan peerstore.PeerInfo
	discovery   *libp2pdis.RoutingDiscovery
	messageChan chan *msg_pb.Message
	started     bool
}

// ConnectionRetry set the number of retry of connection to bootnode in case the initial connection is failed
var (
	// retry for 10 minutes and give up then
	ConnectionRetry = 300
)

const (
	waitInRetry       = 2 * time.Second
	connectionTimeout = 3 * time.Minute

	// register to bootnode every ticker
	dhtTicker = 6 * time.Hour
)

// New returns role conversion service.
func New(h p2p.Host, rendezvous p2p.GroupID, peerChan chan p2p.Peer, bootnodes utils.AddrList) *Service {
	ctx, cancel := context.WithTimeout(context.Background(), connectionTimeout)
	dht, err := libp2pdht.New(ctx, h.GetP2PHost())
	if err != nil {
		panic(err)
	}

	return &Service{
		Host:        h,
		dht:         dht,
		Rendezvous:  rendezvous,
		ctx:         ctx,
		cancel:      cancel,
		stopChan:    make(chan struct{}),
		stoppedChan: make(chan struct{}),
		peerChan:    peerChan,
		bootnodes:   bootnodes,
		discovery:   nil,
		started:     false,
	}
}

// StartService starts network info service.
func (s *Service) StartService() {
	err := s.Init()
	if err != nil {
		utils.GetLogInstance().Error("Service Init Failed", "error", err)
		return
	}
	s.Run()
	s.started = true
}

// Init initializes role conversion service.
func (s *Service) Init() error {
	utils.GetLogInstance().Info("Init networkinfo service")

	// Bootstrap the DHT. In the default configuration, this spawns a Background
	// thread that will refresh the peer table every five minutes.
	utils.GetLogInstance().Debug("Bootstrapping the DHT")
	if err := s.dht.Bootstrap(s.ctx); err != nil {
		return fmt.Errorf("error bootstrap dht")
	}

	var wg sync.WaitGroup
	if s.bootnodes == nil {
		s.bootnodes = utils.BootNodes
	}

	connected := false
	for _, peerAddr := range s.bootnodes {
		peerinfo, _ := peerstore.InfoFromP2pAddr(peerAddr)
		wg.Add(1)
		go func() {
			defer wg.Done()
			for i := 0; i < ConnectionRetry; i++ {
				if err := s.Host.GetP2PHost().Connect(s.ctx, *peerinfo); err != nil {
					utils.GetLogInstance().Warn("can't connect to bootnode", "error", err, "try", i)
					time.Sleep(waitInRetry)
				} else {
					utils.GetLogInstance().Info("connected to bootnode", "node", *peerinfo, "try", i)
					// it is okay if any bootnode is connected
					connected = true
					break
				}
			}
		}()
	}
	wg.Wait()

	if !connected {
		return fmt.Errorf("[FATAL] error connecting to bootnodes")
	}

	// We use a rendezvous point "shardID" to announce our location.
	utils.GetLogInstance().Info("Announcing ourselves...")
	s.discovery = libp2pdis.NewRoutingDiscovery(s.dht)
	libp2pdis.Advertise(s.ctx, s.discovery, string(s.Rendezvous))
	utils.GetLogInstance().Info("Successfully announced!")

	return nil
}

// Run runs network info.
func (s *Service) Run() {
	defer close(s.stoppedChan)
	if s.discovery == nil {
		utils.GetLogInstance().Error("discovery is not initialized")
		return
	}

	var err error
	s.peerInfo, err = s.discovery.FindPeers(s.ctx, string(s.Rendezvous))
	if err != nil {
		utils.GetLogInstance().Error("FindPeers", "error", err)
		return
	}

	go s.DoService()
}

// DoService does network info.
func (s *Service) DoService() {
	_, cgnPrefix, err := net.ParseCIDR("100.64.0.0/10")
	if err != nil {
		utils.GetLogInstance().Error("can't parse CIDR", "error", err)
		return
	}
	tick := time.NewTicker(dhtTicker)
	for {
		select {
		case peer := <-s.peerInfo:
			if peer.ID != s.Host.GetP2PHost().ID() && len(peer.ID) > 0 {
				//	utils.GetLogInstance().Info("Found Peer", "peer", peer.ID, "addr", peer.Addrs, "my ID", s.Host.GetP2PHost().ID())
				if err := s.Host.GetP2PHost().Connect(s.ctx, peer); err != nil {
					utils.GetLogInstance().Warn("can't connect to peer node", "error", err, "peer", peer)
					// break if the node can't connect to peers, waiting for another peer
					break
				} else {
					utils.GetLogInstance().Info("connected to peer node", "peer", peer)
				}
				// figure out the public ip/port
				var ip, port string

				for _, addr := range peer.Addrs {
					netaddr, err := manet.ToNetAddr(addr)
					if err != nil {
						continue
					}
					nip := netaddr.(*net.TCPAddr).IP
					if (nip.IsGlobalUnicast() && !utils.IsPrivateIP(nip)) || cgnPrefix.Contains(nip) {
						ip = nip.String()
						port = fmt.Sprintf("%d", netaddr.(*net.TCPAddr).Port)
						break
					}
				}
				p := p2p.Peer{IP: ip, Port: port, PeerID: peer.ID, Addrs: peer.Addrs}
				utils.GetLogInstance().Info("Notify peerChan", "peer", p)
				if s.peerChan != nil {
					s.peerChan <- p
				}
			}
		case <-s.stopChan:
			return
		case <-tick.C:
			libp2pdis.Advertise(s.ctx, s.discovery, string(s.Rendezvous))
			utils.GetLogInstance().Info("Successfully announced!")
		}
	}
}

// StopService stops network info service.
func (s *Service) StopService() {
	utils.GetLogInstance().Info("Stopping network info service.")
	defer s.cancel()

	if !s.started {
		utils.GetLogInstance().Info("Service didn't started. Exit.")
		return
	}

	s.stopChan <- struct{}{}
	<-s.stoppedChan
	utils.GetLogInstance().Info("Network info service stopped.")
}

// NotifyService notify service
func (s *Service) NotifyService(params map[string]interface{}) {
	return
}

// SetMessageChan sets up message channel to service.
func (s *Service) SetMessageChan(messageChan chan *msg_pb.Message) {
	s.messageChan = messageChan
}