diff --git a/router/connection.go b/router/connection.go index 8eb1625f..08c742c9 100644 --- a/router/connection.go +++ b/router/connection.go @@ -167,7 +167,7 @@ func (conn *LocalConnection) log(args ...interface{}) { // ACTOR client API const ( - CSendTCP = iota + CSendProtocolMsg = iota CSetEstablished CReceivedHeartbeat CShutdown @@ -200,10 +200,10 @@ func (conn *LocalConnection) SetEstablished() { } // Async -func (conn *LocalConnection) SendTCP(msg []byte) { +func (conn *LocalConnection) SendProtocolMsg(tag ProtocolMsg, msg []byte) { conn.queryChan <- &ConnectionInteraction{ - Interaction: Interaction{code: CSendTCP}, - payload: msg} + Interaction: Interaction{code: CSendProtocolMsg}, + payload: TaggedProtocolMsg{tag, msg}} } // ACTOR server @@ -265,16 +265,16 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction) err = conn.handleReceivedHeartbeat(query.payload.(*net.UDPAddr)) case CSetEstablished: err = conn.handleSetEstablished() - case CSendTCP: - err = conn.handleSendTCP(query.payload.([]byte)) + case CSendProtocolMsg: + err = conn.handleSendProtocolMsg(query.payload.(TaggedProtocolMsg)) } case <-tickerChan(conn.heartbeat): conn.Forward(true, conn.heartbeatFrame, nil) case <-tickerChan(conn.fetchAll): - err = conn.handleSendTCP(ProtocolFetchAllByte) + err = conn.handleSendSimpleProtocolMsg(ProtocolFetchAll) case <-tickerChan(conn.fragTest): conn.setStackFrag(false) - err = conn.handleSendTCP(ProtocolStartFragmentationTestByte) + err = conn.handleSendSimpleProtocolMsg(ProtocolStartFragmentationTest) } } return @@ -295,7 +295,7 @@ func (conn *LocalConnection) handleReceivedHeartbeat(remoteUDPAddr *net.UDPAddr) conn.receivedHeartbeat = true conn.Unlock() if !old { - if err := conn.handleSendTCP(ProtocolConnectionEstablishedByte); err != nil { + if err := conn.handleSendSimpleProtocolMsg(ProtocolConnectionEstablished); err != nil { return err } } @@ -332,18 +332,22 @@ func (conn *LocalConnection) handleSetEstablished() error { conn.fragTest = time.NewTicker(FragTestInterval) // avoid initial waits for timers to fire conn.Forward(true, conn.heartbeatFrame, nil) - if err := conn.handleSendTCP(ProtocolFetchAllByte); err != nil { + if err := conn.handleSendSimpleProtocolMsg(ProtocolFetchAll); err != nil { return err } conn.setStackFrag(false) - if err := conn.handleSendTCP(ProtocolStartFragmentationTestByte); err != nil { + if err := conn.handleSendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil { return err } return nil } -func (conn *LocalConnection) handleSendTCP(msg []byte) error { - return conn.tcpSender.Send(msg) +func (conn *LocalConnection) handleSendSimpleProtocolMsg(tag ProtocolMsg) error { + return conn.handleSendProtocolMsg(TaggedProtocolMsg{tag: tag}) +} + +func (conn *LocalConnection) handleSendProtocolMsg(m TaggedProtocolMsg) error { + return conn.tcpSender.Send(Concat([]byte{byte(m.tag)}, m.msg)) } func (conn *LocalConnection) handleShutdown() { @@ -564,13 +568,13 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder) { // payload is a subset of the receiver's topology, no // further action is taken. Otherwise, the receiver sends // out to all its connections an "improved" update. - conn.SendTCP(Concat(ProtocolUpdateByte, conn.Router.Peers.EncodeAllPeers())) + conn.SendProtocolMsg(ProtocolUpdate, conn.Router.Peers.EncodeAllPeers()) case ProtocolUpdate: newUpdate, err := conn.Router.Peers.ApplyUpdate(payload) if _, ok := err.(UnknownPeersError); err != nil && ok { // That update contained a peer we didn't know about; // request full update - conn.SendTCP(ProtocolFetchAllByte) + conn.SendProtocolMsg(ProtocolFetchAll, nil) continue } if conn.CheckFatal(err) != nil { @@ -579,7 +583,7 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder) { if len(newUpdate) != 0 { conn.Router.ConnectionMaker.Refresh() conn.Router.Routes.Recalculate() - conn.Router.Ourself.BroadcastTCP(Concat(ProtocolUpdateByte, newUpdate)) + conn.Router.Ourself.SendProtocolMsg(ProtocolUpdate, newUpdate) } case ProtocolPMTUVerified: conn.verifyPMTU <- int(binary.BigEndian.Uint16(payload)) diff --git a/router/crypto.go b/router/crypto.go index ef9c1b4b..0bb7dd14 100644 --- a/router/crypto.go +++ b/router/crypto.go @@ -85,7 +85,7 @@ func EncodeNonce(df bool) (*[24]byte, []byte, error) { flags = flags | 1 } SetNonceLow15Bits(&nonce, flags) - return &nonce, Concat(ProtocolNonceByte, nonce[:]), nil + return &nonce, nonce[:], nil } func DecodeNonce(msg []byte) (bool, *[24]byte) { @@ -206,7 +206,7 @@ func (ne *NaClEncryptor) Bytes() []byte { if err = ne.conn.CheckFatal(err); err != nil { return []byte{} } - ne.conn.SendTCP(encodedNonce) + ne.conn.SendProtocolMsg(ProtocolNonce, encodedNonce) ne.nonce = freshNonce nonce = freshNonce } @@ -225,7 +225,7 @@ func (ne *NaClEncryptor) Bytes() []byte { return []byte{} } ne.nonceChan <- nonce - ne.conn.SendTCP(encodedNonce) + ne.conn.SendProtocolMsg(ProtocolNonce, encodedNonce) } ne.offset = offset diff --git a/router/local_peer.go b/router/local_peer.go index 9c85959f..62bbeb7b 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -117,9 +117,9 @@ func (peer *LocalPeer) CreateConnection(peerAddr string, acceptNewPeer bool) err const ( PAddConnection = iota - PBroadcastTCP PDeleteConnection PConnectionEstablished + PSendProtocolMsg ) // Async: rely on the peer to shut us down if we shouldn't be adding @@ -147,10 +147,10 @@ func (peer *LocalPeer) ConnectionEstablished(conn *LocalConnection) { } // Async. -func (peer *LocalPeer) BroadcastTCP(msg []byte) { +func (peer *LocalPeer) SendProtocolMsg(tag ProtocolMsg, msg []byte) { peer.queryChan <- &PeerInteraction{ - Interaction: Interaction{code: PBroadcastTCP}, - payload: msg} + Interaction: Interaction{code: PSendProtocolMsg}, + payload: TaggedProtocolMsg{tag, msg}} } // ACTOR server @@ -169,8 +169,8 @@ func (peer *LocalPeer) queryLoop(queryChan <-chan *PeerInteraction) { query.resultChan <- nil case PConnectionEstablished: peer.handleConnectionEstablished(query.payload.(*LocalConnection)) - case PBroadcastTCP: - peer.handleBroadcastTCP(query.payload.([]byte)) + case PSendProtocolMsg: + peer.handleSendProtocolMsg(query.payload.(TaggedProtocolMsg)) } } } @@ -248,9 +248,9 @@ func (peer *LocalPeer) handleConnectionEstablished(conn *LocalConnection) { peer.broadcastPeerUpdate(conn.Remote()) } -func (peer *LocalPeer) handleBroadcastTCP(msg []byte) { +func (peer *LocalPeer) handleSendProtocolMsg(m TaggedProtocolMsg) { peer.ForEachConnection(func(_ PeerName, conn Connection) { - conn.(*LocalConnection).SendTCP(msg) + conn.(*LocalConnection).SendProtocolMsg(m.tag, m.msg) }) } @@ -278,7 +278,8 @@ func (peer *LocalPeer) connectionEstablished(conn Connection) { func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) { peer.Router.Routes.Recalculate() - peer.handleBroadcastTCP(Concat(ProtocolUpdateByte, EncodePeers(append(peers, peer.Peer)...))) + update := EncodePeers(append(peers, peer.Peer)...) + peer.handleSendProtocolMsg(TaggedProtocolMsg{tag: ProtocolUpdate, msg: update}) } func (peer *LocalPeer) checkConnectionLimit() error { diff --git a/router/protocol.go b/router/protocol.go index 9fffe7f5..df22d918 100644 --- a/router/protocol.go +++ b/router/protocol.go @@ -7,6 +7,11 @@ const ( type ProtocolMsg byte +type TaggedProtocolMsg struct { + tag ProtocolMsg + msg []byte +} + const ( ProtocolConnectionEstablished ProtocolMsg = iota ProtocolFragmentationReceived @@ -16,13 +21,3 @@ const ( ProtocolUpdate ProtocolPMTUVerified ) - -var ( - ProtocolConnectionEstablishedByte = []byte{byte(ProtocolConnectionEstablished)} - ProtocolFragmentationReceivedByte = []byte{byte(ProtocolFragmentationReceived)} - ProtocolStartFragmentationTestByte = []byte{byte(ProtocolStartFragmentationTest)} - ProtocolNonceByte = []byte{byte(ProtocolNonce)} - ProtocolFetchAllByte = []byte{byte(ProtocolFetchAll)} - ProtocolUpdateByte = []byte{byte(ProtocolUpdate)} - ProtocolPMTUVerifiedByte = []byte{byte(ProtocolPMTUVerified)} -) diff --git a/router/router.go b/router/router.go index d9e31687..33c03f51 100644 --- a/router/router.go +++ b/router/router.go @@ -308,12 +308,12 @@ func (router *Router) handleUDPPacketFunc(dec *EthernetDecoder, po PacketSink) F case frameLen == EthernetOverhead+8: relayConn.ReceivedHeartbeat(sender, binary.BigEndian.Uint64(frame[EthernetOverhead:])) case frameLen == FragTestSize && bytes.Equal(frame, FragTest): - relayConn.SendTCP(ProtocolFragmentationReceivedByte) + relayConn.SendProtocolMsg(ProtocolFragmentationReceived, nil) case frameLen == PMTUDiscoverySize && bytes.Equal(frame, PMTUDiscovery): default: frameLenBytes := []byte{0, 0} binary.BigEndian.PutUint16(frameLenBytes, uint16(frameLen-EthernetOverhead)) - relayConn.SendTCP(Concat(ProtocolPMTUVerifiedByte, frameLenBytes)) + relayConn.SendProtocolMsg(ProtocolPMTUVerified, frameLenBytes) } return nil }