woop/api/service/networkinfo/service.go

package networkinfo

import (
	"context"
	"fmt"
	"net"
	"strings"
	"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
}

// New returns role conversion service.
func New(h p2p.Host, rendezvous p2p.GroupID, peerChan chan p2p.Peer, bootnodes utils.AddrList) *Service {
	timeout := 30 * time.Minute
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	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,
	}
}

// StartService starts network info service.
func (s *Service) StartService() {
	s.Init()
	s.Run()
}

// 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
	}

	for _, peerAddr := range s.bootnodes {
		peerinfo, _ := peerstore.InfoFromP2pAddr(peerAddr)
		wg.Add(1)
		go func() {
			defer wg.Done()
			if err := s.Host.GetP2PHost().Connect(s.ctx, *peerinfo); err != nil {
				utils.GetLogInstance().Warn("can't connect to bootnode", "error", err)
			} else {
				utils.GetLogInstance().Info("connected to bootnode", "node", *peerinfo)
			}
		}()
	}
	wg.Wait()

	// 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)
	var err error
	s.peerInfo, err = s.discovery.FindPeers(s.ctx, string(s.Rendezvous))
	if err != nil {
		utils.GetLogInstance().Error("FindPeers", "error", err)
	}

	go s.DoService()
}

// DoService does network info.
func (s *Service) DoService() {
	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)
				} else {
					utils.GetLogInstance().Info("connected to peer node", "peer", peer)
				}
				// figure out the public ip/port
				ip := "127.0.0.1"
				var port string
				for _, addr := range peer.Addrs {
					netaddr, err := manet.ToNetAddr(addr)
					if err != nil {
						continue
					}
					nip := netaddr.(*net.TCPAddr).IP.String()
					if strings.Compare(nip, "127.0.0.1") != 0 {
						ip = nip
						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
		}
	}
}

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

	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
}