rename and lift ProtocolMsg into API
This commit is contained in:
1 parent
d089900632
commit
ac95d5ab29
5 files changed
+24
-25
No files matched your search
+10
-10
@@ -200,10 +200,10 @@ func (conn *LocalConnection) SetEstablished() {
|
||||
}
|
||||
|
||||
// Async
|
||||
func (conn *LocalConnection) SendProtocolMsg(tag ProtocolMsg, msg []byte) {
|
||||
func (conn *LocalConnection) SendProtocolMsg(m ProtocolMsg) {
|
||||
conn.queryChan <- &ConnectionInteraction{
|
||||
Interaction: Interaction{code: CSendProtocolMsg},
|
||||
payload: TaggedProtocolMsg{tag, msg}}
|
||||
payload: m}
|
||||
}
|
||||
|
||||
// ACTOR server
|
||||
@@ -266,7 +266,7 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction)
|
||||
case CSetEstablished:
|
||||
err = conn.handleSetEstablished()
|
||||
case CSendProtocolMsg:
|
||||
err = conn.handleSendProtocolMsg(query.payload.(TaggedProtocolMsg))
|
||||
err = conn.handleSendProtocolMsg(query.payload.(ProtocolMsg))
|
||||
}
|
||||
case <-tickerChan(conn.heartbeat):
|
||||
conn.Forward(true, conn.heartbeatFrame, nil)
|
||||
@@ -342,11 +342,11 @@ func (conn *LocalConnection) handleSetEstablished() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (conn *LocalConnection) handleSendSimpleProtocolMsg(tag ProtocolMsg) error {
|
||||
return conn.handleSendProtocolMsg(TaggedProtocolMsg{tag: tag})
|
||||
func (conn *LocalConnection) handleSendSimpleProtocolMsg(tag ProtocolTag) error {
|
||||
return conn.handleSendProtocolMsg(ProtocolMsg{tag: tag})
|
||||
}
|
||||
|
||||
func (conn *LocalConnection) handleSendProtocolMsg(m TaggedProtocolMsg) error {
|
||||
func (conn *LocalConnection) handleSendProtocolMsg(m ProtocolMsg) error {
|
||||
return conn.tcpSender.Send(Concat([]byte{byte(m.tag)}, m.msg))
|
||||
}
|
||||
|
||||
@@ -534,7 +534,7 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder) {
|
||||
continue
|
||||
}
|
||||
payload := msg[1:]
|
||||
switch ProtocolMsg(msg[0]) {
|
||||
switch ProtocolTag(msg[0]) {
|
||||
case ProtocolConnectionEstablished:
|
||||
// We sent fast heartbeats to the remote peer, which has
|
||||
// now received at least one of them and told us via this
|
||||
@@ -568,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.SendProtocolMsg(ProtocolUpdate, conn.Router.Peers.EncodeAllPeers())
|
||||
conn.SendProtocolMsg(ProtocolMsg{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.SendProtocolMsg(ProtocolFetchAll, nil)
|
||||
conn.SendProtocolMsg(ProtocolMsg{ProtocolFetchAll, nil})
|
||||
continue
|
||||
}
|
||||
if conn.CheckFatal(err) != nil {
|
||||
@@ -583,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.SendProtocolMsg(ProtocolUpdate, newUpdate)
|
||||
conn.Router.Ourself.SendProtocolMsg(ProtocolMsg{ProtocolUpdate, newUpdate})
|
||||
}
|
||||
case ProtocolPMTUVerified:
|
||||
conn.verifyPMTU <- int(binary.BigEndian.Uint16(payload))
|
||||
|
||||
+2
-2
@@ -207,7 +207,7 @@ func (ne *NaClEncryptor) Bytes() []byte {
|
||||
if err = ne.conn.CheckFatal(err); err != nil {
|
||||
return []byte{}
|
||||
}
|
||||
ne.conn.SendProtocolMsg(ProtocolNonce, encodedNonce)
|
||||
ne.conn.SendProtocolMsg(ProtocolMsg{ProtocolNonce, encodedNonce})
|
||||
ne.nonce = freshNonce
|
||||
nonce = freshNonce
|
||||
}
|
||||
@@ -226,7 +226,7 @@ func (ne *NaClEncryptor) Bytes() []byte {
|
||||
return []byte{}
|
||||
}
|
||||
ne.nonceChan <- nonce
|
||||
ne.conn.SendProtocolMsg(ProtocolNonce, encodedNonce)
|
||||
ne.conn.SendProtocolMsg(ProtocolMsg{ProtocolNonce, encodedNonce})
|
||||
}
|
||||
ne.offset = offset
|
||||
|
||||
|
||||
@@ -147,10 +147,10 @@ func (peer *LocalPeer) ConnectionEstablished(conn *LocalConnection) {
|
||||
}
|
||||
|
||||
// Async.
|
||||
func (peer *LocalPeer) SendProtocolMsg(tag ProtocolMsg, msg []byte) {
|
||||
func (peer *LocalPeer) SendProtocolMsg(m ProtocolMsg) {
|
||||
peer.queryChan <- &PeerInteraction{
|
||||
Interaction: Interaction{code: PSendProtocolMsg},
|
||||
payload: TaggedProtocolMsg{tag, msg}}
|
||||
payload: m}
|
||||
}
|
||||
|
||||
// ACTOR server
|
||||
@@ -170,7 +170,7 @@ func (peer *LocalPeer) queryLoop(queryChan <-chan *PeerInteraction) {
|
||||
case PConnectionEstablished:
|
||||
peer.handleConnectionEstablished(query.payload.(*LocalConnection))
|
||||
case PSendProtocolMsg:
|
||||
peer.handleSendProtocolMsg(query.payload.(TaggedProtocolMsg))
|
||||
peer.handleSendProtocolMsg(query.payload.(ProtocolMsg))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -248,9 +248,9 @@ func (peer *LocalPeer) handleConnectionEstablished(conn *LocalConnection) {
|
||||
peer.broadcastPeerUpdate(conn.Remote())
|
||||
}
|
||||
|
||||
func (peer *LocalPeer) handleSendProtocolMsg(m TaggedProtocolMsg) {
|
||||
func (peer *LocalPeer) handleSendProtocolMsg(m ProtocolMsg) {
|
||||
peer.ForEachConnection(func(_ PeerName, conn Connection) {
|
||||
conn.(*LocalConnection).SendProtocolMsg(m.tag, m.msg)
|
||||
conn.(*LocalConnection).SendProtocolMsg(m)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -278,8 +278,7 @@ func (peer *LocalPeer) connectionEstablished(conn Connection) {
|
||||
|
||||
func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) {
|
||||
peer.Router.Routes.Recalculate()
|
||||
update := EncodePeers(append(peers, peer.Peer)...)
|
||||
peer.handleSendProtocolMsg(TaggedProtocolMsg{tag: ProtocolUpdate, msg: update})
|
||||
peer.handleSendProtocolMsg(ProtocolMsg{ProtocolUpdate, EncodePeers(append(peers, peer.Peer)...)})
|
||||
}
|
||||
|
||||
func (peer *LocalPeer) checkConnectionLimit() error {
|
||||
|
||||
+4
-4
@@ -5,15 +5,15 @@ const (
|
||||
ProtocolVersion = 10
|
||||
)
|
||||
|
||||
type ProtocolMsg byte
|
||||
type ProtocolTag byte
|
||||
|
||||
type TaggedProtocolMsg struct {
|
||||
tag ProtocolMsg
|
||||
type ProtocolMsg struct {
|
||||
tag ProtocolTag
|
||||
msg []byte
|
||||
}
|
||||
|
||||
const (
|
||||
ProtocolConnectionEstablished ProtocolMsg = iota
|
||||
ProtocolConnectionEstablished ProtocolTag = iota
|
||||
ProtocolFragmentationReceived
|
||||
ProtocolStartFragmentationTest
|
||||
ProtocolNonce
|
||||
|
||||
+2
-2
@@ -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.SendProtocolMsg(ProtocolFragmentationReceived, nil)
|
||||
relayConn.SendProtocolMsg(ProtocolMsg{ProtocolFragmentationReceived, nil})
|
||||
case frameLen == PMTUDiscoverySize && bytes.Equal(frame, PMTUDiscovery):
|
||||
default:
|
||||
frameLenBytes := []byte{0, 0}
|
||||
binary.BigEndian.PutUint16(frameLenBytes, uint16(frameLen-EthernetOverhead))
|
||||
relayConn.SendProtocolMsg(ProtocolPMTUVerified, frameLenBytes)
|
||||
relayConn.SendProtocolMsg(ProtocolMsg{ProtocolPMTUVerified, frameLenBytes})
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user