permit encoding of GossipData into multiple messages

This commit is contained in:
Matthias Radestock committed 2015-06-15 11:14:24 +01:00
1 parent 94220f22b2
commit 0b3584a355
5 files changed
+20 -13

No files matched your search

+2 -2
View File
@@ -435,8 +435,8 @@ func (d *ipamGossipData) Merge(other router.GossipData) {
// no-op
}
func (d *ipamGossipData) Encode() []byte {
return d.alloc.Encode()
func (d *ipamGossipData) Encode() [][]byte {
return [][]byte{d.alloc.Encode()}
}
// Gossip returns a GossipData implementation, which in this case always
+4 -2
View File
@@ -258,8 +258,10 @@ func (grouter *TestGossipRouter) connect(sender router.PeerName, gossiper router
panic(fmt.Sprintf("Error doing gossip unicast to %s: %s", sender, err))
}
} else {
if _, err := gossiper.OnGossipBroadcast(message.data.Encode()); err != nil {
panic(fmt.Sprintf("Error doing gossip broadcast to %s: %s", sender, err))
for _, msg := range message.data.Encode() {
if _, err := gossiper.OnGossipBroadcast(msg); err != nil {
panic(fmt.Sprintf("Error doing gossip broadcast to %s: %s", sender, err))
}
}
}
case <-gossipTimer:
+1 -1
View File
@@ -5,7 +5,7 @@ import (
)
type GossipData interface {
Encode() []byte
Encode() [][]byte
Merge(GossipData)
}
+11 -6
View File
@@ -145,8 +145,10 @@ func (c *GossipChannel) sendDown(conn Connection, data GossipData) {
sender, found := c.senders[conn]
if !found {
sender = NewGossipSender(func(pending GossipData) {
protocolMsg := ProtocolMsg{ProtocolGossip, GobEncode(c.name, c.ourself.Name, pending.Encode())}
conn.(ProtocolSender).SendProtocolMsg(protocolMsg)
for _, msg := range pending.Encode() {
protocolMsg := ProtocolMsg{ProtocolGossip, GobEncode(c.name, c.ourself.Name, msg)}
conn.(ProtocolSender).SendProtocolMsg(protocolMsg)
}
})
c.senders[conn] = sender
sender.Start()
@@ -212,10 +214,13 @@ func (c *GossipChannel) sendBroadcast(srcName PeerName, update GossipData) {
if len(nextHops) == 0 {
return
}
protocolMsg := ProtocolMsg{ProtocolGossipBroadcast, GobEncode(c.name, srcName, update.Encode())}
// FIXME a single blocked connection can stall us
for _, conn := range c.ourself.ConnectionsTo(nextHops) {
conn.(ProtocolSender).SendProtocolMsg(protocolMsg)
connections := c.ourself.ConnectionsTo(nextHops)
for _, msg := range update.Encode() {
protocolMsg := ProtocolMsg{ProtocolGossipBroadcast, GobEncode(c.name, srcName, msg)}
// FIXME a single blocked connection can stall us
for _, conn := range connections {
conn.(ProtocolSender).SendProtocolMsg(protocolMsg)
}
}
}
+2 -2
View File
@@ -394,8 +394,8 @@ func (d *TopologyGossipData) Merge(other GossipData) {
}
}
func (d *TopologyGossipData) Encode() []byte {
return d.peers.EncodePeers(d.update)
func (d *TopologyGossipData) Encode() [][]byte {
return [][]byte{d.peers.EncodePeers(d.update)}
}
func (router *Router) OnGossipUnicast(sender PeerName, msg []byte) error {