137 lines
2.9 KiB
Go
137 lines
2.9 KiB
Go
package node
|
|
|
|
import (
|
|
"bytes"
|
|
"lux/crypto"
|
|
"lux/host"
|
|
"lux/net"
|
|
"lux/proto"
|
|
"sync"
|
|
|
|
ipnet "net"
|
|
)
|
|
|
|
type LuxNode struct {
|
|
nodeId *proto.LuxID
|
|
router net.LuxRouter
|
|
running bool
|
|
|
|
neighbors map[proto.LuxID]*ipnet.UDPAddr
|
|
|
|
state LuxNodeState
|
|
stateLock sync.RWMutex
|
|
}
|
|
|
|
func NewLuxNode(nodeKey crypto.LuxKey, ks crypto.LuxKeyStore) LuxNode {
|
|
return LuxNode{
|
|
nodeId: &nodeKey.Id,
|
|
router: net.NewLuxRouter(nodeKey, ks),
|
|
running: false,
|
|
neighbors: make(map[proto.LuxID]*ipnet.UDPAddr),
|
|
state: NewNodeState(),
|
|
}
|
|
}
|
|
|
|
func (node *LuxNode) GetRouter() *net.LuxRouter {
|
|
return &node.router
|
|
}
|
|
|
|
func (node *LuxNode) GetNodeID() proto.LuxID {
|
|
return *node.nodeId
|
|
}
|
|
|
|
func (node *LuxNode) AddInterior(udpAddr string) error {
|
|
return node.router.CreateInboundChannel(net.LuxChannelInterior, udpAddr)
|
|
}
|
|
|
|
func (node *LuxNode) AddExterior(udpAddr string) error {
|
|
return node.router.CreateInboundChannel(net.LuxChannelExterior, udpAddr)
|
|
}
|
|
|
|
func (node *LuxNode) AddNeighbor(id proto.LuxID, udpAddr string) error {
|
|
if bytes.Equal(id.UUID[:], node.nodeId.UUID[:]) {
|
|
log.Debug("skipping neighbor pointing to this node %s", node.nodeId.String())
|
|
return nil
|
|
}
|
|
|
|
udpIp, err := ipnet.ResolveUDPAddr("udp", udpAddr)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if node.router.HasKeyFor(id) {
|
|
// we have key for this node, so we can route
|
|
err = node.router.CreateOutboundRoute(id, net.LuxChannelInterior, udpAddr)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
// just add to neighbor list to let other nodes know
|
|
node.neighbors[id] = udpIp
|
|
|
|
log.Infof("added neighbor %s at %s", id.String(), udpIp.String())
|
|
return nil
|
|
}
|
|
|
|
func (node *LuxNode) GetState() *LuxNodeState {
|
|
return &node.state
|
|
}
|
|
|
|
func (node *LuxNode) GetStateLock() *sync.RWMutex {
|
|
return &node.stateLock
|
|
}
|
|
|
|
func (node *LuxNode) GetHostStateChannel() <-chan LuxHostState {
|
|
return node.state.GetStateChannel()
|
|
}
|
|
|
|
func (node *LuxNode) handleHeartbeat(packet *net.LuxPacket) {
|
|
if packet.ChannelType != net.LuxChannelExterior {
|
|
log.Error("heartbeat not on exterior!")
|
|
return
|
|
}
|
|
|
|
// parse heartbeat
|
|
state := host.NewLuxState("")
|
|
if err := state.Read(&packet.Buffer); err != nil {
|
|
log.Error("failed to parse heartbeat: %v", err)
|
|
return
|
|
}
|
|
|
|
// register heartbeat
|
|
node.stateLock.Lock()
|
|
node.state.RegisterHeartbeat(packet.Target, state)
|
|
node.stateLock.Unlock()
|
|
|
|
log.Debugf("heartbeat from %s", packet.Target.String())
|
|
}
|
|
|
|
func nodeLoop(node *LuxNode) {
|
|
router := &node.router
|
|
for node.running {
|
|
packet, err := router.Recv()
|
|
if err != nil {
|
|
log.Infof("node recv err %v", err)
|
|
}
|
|
|
|
switch packet.Type {
|
|
case net.LuxPacketTypeHeartbeat:
|
|
node.handleHeartbeat(&packet)
|
|
default:
|
|
log.Warningf("unknown packet type %d on chType %d", packet.Type, packet.ChannelType)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (node *LuxNode) Start() {
|
|
node.router.Start()
|
|
node.running = true
|
|
|
|
go nodeLoop(node)
|
|
}
|
|
|
|
func (node *LuxNode) Stop() {
|
|
node.running = false
|
|
node.router.Stop()
|
|
}
|