Add basic integration tests for the router

cmd/pineconesim/simulator/adversary/drop_packets.go

@@ -119,8 +119,8 @@ func NewAdversaryRouter(log *log.Logger, sk ed25519.PrivateKey, debug bool) *Adv
 	return adversary
 }
 
-func (a *AdversaryRouter) Subscribe(ch chan events.Event) {
-	a.rtr.Subscribe(ch)
+func (a *AdversaryRouter) Subscribe(ch chan events.Event) router.NodeState {
+	return a.rtr.Subscribe(ch)
 }
 
 func (a *AdversaryRouter) PublicKey() types.PublicKey {

cmd/pineconesim/simulator/nodes.go

@@ -127,14 +127,14 @@ func (sim *Simulator) StartNodeEventHandler(t string, nodeType APINodeType) {
 	ch := make(chan events.Event)
 	handler := eventHandler{node: t, ch: ch}
 	quit := make(chan bool)
-	go handler.Run(quit, sim)
-	sim.nodes[t].Subscribe(ch)
+	nodeState := sim.nodes[t].Subscribe(ch)
 
 	sim.nodeRunnerChannelsMutex.Lock()
 	sim.nodeRunnerChannels[t] = append(sim.nodeRunnerChannels[t], quit)
 	sim.nodeRunnerChannelsMutex.Unlock()
 
-	phony.Block(sim.State, func() { sim.State._addNode(t, sim.nodes[t].PublicKey().String(), nodeType) })
+	phony.Block(sim.State, func() { sim.State._addNode(t, sim.nodes[t].PublicKey().String(), nodeType, nodeState) })
+	go handler.Run(quit, sim)
 }
 
 func (sim *Simulator) RemoveNode(node string) {

cmd/pineconesim/simulator/router.go

@@ -31,7 +31,7 @@ import (
 type SimRouter interface {
 	PublicKey() types.PublicKey
 	Connect(conn net.Conn, options ...router.ConnectionOption) (types.SwitchPortID, error)
-	Subscribe(ch chan events.Event)
+	Subscribe(ch chan events.Event) router.NodeState
 	Ping(ctx context.Context, a net.Addr) (uint16, time.Duration, error)
 	Coords() types.Coordinates
 	ConfigureFilterDefaults(rates adversary.DropRates)
@@ -44,8 +44,8 @@ type DefaultRouter struct {
 	pings sync.Map // types.PublicKey -> chan struct{}
 }
 
-func (r *DefaultRouter) Subscribe(ch chan events.Event) {
-	r.rtr.Subscribe(ch)
+func (r *DefaultRouter) Subscribe(ch chan events.Event) router.NodeState {
+	return r.rtr.Subscribe(ch)
 }
 
 func (r *DefaultRouter) PublicKey() types.PublicKey {

cmd/pineconesim/simulator/simulator.go

@@ -382,6 +382,8 @@ func (sim *Simulator) handleTreeRootAnnUpdate(node string, root string, sequence
 	rootName := ""
 	if peerNode, err := sim.State.GetNodeName(root); err == nil {
 		rootName = peerNode
+	} else {
+		log.Fatalf("Cannot convert %s to root for %s", root, node)
 	}
 	sim.State.Act(nil, func() { sim.State._updateTreeRootAnnouncement(node, rootName, sequence, time, coords) })
 }

cmd/pineconesim/simulator/state.go

@@ -19,6 +19,7 @@ import (
 	"reflect"
 
 	"github.com/Arceliar/phony"
+	"github.com/matrix-org/pinecone/router"
 )
 
 type RootAnnouncement struct {
@@ -174,13 +175,7 @@ func (s *StateAccessor) GetNodeName(peerID string) (string, error) {
 	node := ""
 	err := fmt.Errorf("Provided peerID is not associated with a known node")
 
-	phony.Block(s, func() {
-		for k, v := range s._state.Nodes {
-			if v.PeerID == peerID {
-				node, err = k, nil
-			}
-		}
-	})
+	phony.Block(s, func() { node, err = s._getNodeName(peerID) })
 	return node, err
 }
 
@@ -195,8 +190,54 @@ func (s *StateAccessor) GetNodeCoords(name string) []uint64 {
 	return coords
 }
 
-func (s *StateAccessor) _addNode(name string, peerID string, nodeType APINodeType) {
+func (s *StateAccessor) _getNodeName(peerID string) (string, error) {
+	node := ""
+	err := fmt.Errorf("Provided peerID is not associated with a known node")
+
+	for k, v := range s._state.Nodes {
+		if v.PeerID == peerID {
+			node, err = k, nil
+		}
+	}
+
+	return node, err
+}
+
+func (s *StateAccessor) _addNode(name string, peerID string, nodeType APINodeType, nodeState router.NodeState) {
 	s._state.Nodes[name] = NewNodeState(peerID, nodeType)
+	if peernode, err := s._getNodeName(nodeState.Parent); err == nil {
+		s._state.Nodes[name].Parent = peernode
+	}
+	connections := map[int]string{}
+	for i, node := range nodeState.Connections {
+		if i == 0 {
+			// NOTE : Skip connection on port 0 since it is the loopback port
+			continue
+		}
+		if peernode, err := s._getNodeName(node); err == nil {
+			connections[i] = peernode
+		}
+	}
+	s._state.Nodes[name].Connections = connections
+	s._state.Nodes[name].Coords = nodeState.Coords
+	root := ""
+	if peernode, err := s._getNodeName(nodeState.Announcement.RootPublicKey.String()); err == nil {
+		root = peernode
+	}
+	announcement := RootAnnouncement{
+		Root:     root,
+		Sequence: uint64(nodeState.Announcement.RootSequence),
+		Time:     nodeState.AnnouncementTime,
+	}
+	s._state.Nodes[name].Announcement = announcement
+	if peernode, err := s._getNodeName(nodeState.AscendingPeer); err == nil {
+		s._state.Nodes[name].AscendingPeer = peernode
+	}
+	s._state.Nodes[name].AscendingPathID = nodeState.AscendingPathID
+	if peernode, err := s._getNodeName(nodeState.DescendingPeer); err == nil {
+		s._state.Nodes[name].DescendingPeer = peernode
+	}
+	s._state.Nodes[name].DescendingPathID = nodeState.DescendingPathID
 	s._publish(NodeAdded{Node: name, PublicKey: peerID, NodeType: int(nodeType)})
 }
 

router/api.go

@@ -38,11 +38,64 @@ type PeerInfo struct {
 	Zone      string
 }
 
+type NodeState struct {
+	PeerID           string
+	Connections      map[int]string
+	Parent           string
+	Coords           []uint64
+	Announcement     types.SwitchAnnouncement
+	AnnouncementTime uint64
+	AscendingPeer    string
+	AscendingPathID  string
+	DescendingPeer   string
+	DescendingPathID string
+}
+
 // Subscribe registers a subscriber to this node's events
-func (r *Router) Subscribe(ch chan<- events.Event) {
+func (r *Router) Subscribe(ch chan<- events.Event) NodeState {
+	var stateCopy NodeState
 	phony.Block(r, func() {
 		r._subscribers[ch] = &phony.Inbox{}
+		stateCopy.PeerID = r.public.String()
+		connections := map[int]string{}
+		for _, p := range r.state._peers {
+			if p == nil {
+				continue
+			}
+			connections[int(p.port)] = p.public.String()
+		}
+		stateCopy.Connections = connections
+		parent := ""
+		if r.state._parent != nil {
+			parent = r.state._parent.public.String()
+		}
+		stateCopy.Parent = parent
+		coords := []uint64{}
+		for _, coord := range r.Coords() {
+			coords = append(coords, uint64(coord))
+		}
+		stateCopy.Coords = coords
+		announcement := r.state._rootAnnouncement()
+		stateCopy.Announcement = announcement.SwitchAnnouncement
+		stateCopy.AnnouncementTime = uint64(announcement.receiveTime.UnixNano())
+		asc := ""
+		ascPath := ""
+		if r.state._ascending != nil {
+			asc = r.state._ascending.PublicKey.String()
+			ascPath = hex.EncodeToString(r.state._ascending.PathID[:])
+		}
+		stateCopy.AscendingPeer = asc
+		stateCopy.AscendingPathID = ascPath
+		desc := ""
+		descPath := ""
+		if r.state._descending != nil {
+			desc = r.state._descending.PublicKey.String()
+			descPath = hex.EncodeToString(r.state._descending.PathID[:])
+		}
+		stateCopy.DescendingPeer = desc
+		stateCopy.DescendingPathID = descPath
 	})
+	return stateCopy
 }
 
 func (r *Router) Coords() types.Coordinates {