reweave/weave
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

better APIs for sending protocol messages

Browse Source 
no longer tagging requiring the caller to construct them
This commit is contained in:
Matthias Radestock committed 2015-01-20 22:31:16 +00:00
1 parent 644697977b
commit 1181a0cb33
5 files changed
+40 -40
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

router/connection.go
+20 -16
View File
@@ -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))
router/crypto.go
+3 -3
View File
@@ -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
router/local_peer.go
+10 -9
View File
@@ -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 {
router/protocol.go
+5 -10
View File
@@ -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)}
)
router/router.go
+2 -2
View File
@@ -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
}