make gossip broadcast operate on GossipData

This commit is contained in:
Matthias Radestock committed 2015-04-06 14:00:51 +01:00
1 parent 9b9d8a03aa
commit a58eb6ba7d
2 files changed
+15 -12

No files matched your search

+13 -10
View File
@@ -20,13 +20,15 @@ type Gossip interface {
// specific message from one peer to another
// intermediate peers relay it using unicast topology.
GossipUnicast(dstPeerName PeerName, msg []byte) error
// send a message to every peer, relayed using broadcast topology.
GossipBroadcast(msg []byte) error
// send gossip to every peer, relayed using broadcast topology.
GossipBroadcast(update GossipData) error
}
type Gossiper interface {
OnGossipUnicast(sender PeerName, msg []byte) error
OnGossipBroadcast(msg []byte) error
// merge received data into state and return a representation of
// the received data, for further propagation
OnGossipBroadcast(update []byte) (GossipData, error)
// return state of everything we know; gets called periodically
Gossip() GossipData
// merge received data into state and return "everything new I've
@@ -150,15 +152,16 @@ func (c *GossipChannel) deliverUnicast(srcName PeerName, origPayload []byte, dec
return c.gossiper.OnGossipUnicast(srcName, payload)
}
func (c *GossipChannel) deliverBroadcast(srcName PeerName, origPayload []byte, dec *gob.Decoder) error {
func (c *GossipChannel) deliverBroadcast(srcName PeerName, _ []byte, dec *gob.Decoder) error {
var payload []byte
if err := dec.Decode(&payload); err != nil {
return err
}
if err := c.gossiper.OnGossipBroadcast(payload); err != nil {
data, err := c.gossiper.OnGossipBroadcast(payload)
if err != nil || data == nil {
return err
}
return c.relayBroadcast(srcName, origPayload)
return c.relayBroadcast(srcName, data)
}
func (c *GossipChannel) deliver(_ PeerName, _ []byte, dec *gob.Decoder) error {
@@ -214,8 +217,8 @@ func (c *GossipChannel) GossipUnicast(dstPeerName PeerName, msg []byte) error {
return c.relayUnicast(dstPeerName, GobEncode(c.hash, c.ourself.Name, dstPeerName, msg))
}
func (c *GossipChannel) GossipBroadcast(msg []byte) error {
return c.relayBroadcast(c.ourself.Name, GobEncode(c.hash, c.ourself.Name, msg))
func (c *GossipChannel) GossipBroadcast(update GossipData) error {
return c.relayBroadcast(c.ourself.Name, update)
}
func (c *GossipChannel) relayUnicast(dstPeerName PeerName, buf []byte) error {
@@ -229,12 +232,12 @@ func (c *GossipChannel) relayUnicast(dstPeerName PeerName, buf []byte) error {
return nil
}
func (c *GossipChannel) relayBroadcast(srcName PeerName, buf []byte) error {
func (c *GossipChannel) relayBroadcast(srcName PeerName, update GossipData) error {
nextHops := c.ourself.Router.Routes.BroadcastAll(srcName)
if len(nextHops) == 0 {
return nil
}
protocolMsg := ProtocolMsg{ProtocolGossipBroadcast, buf}
protocolMsg := ProtocolMsg{ProtocolGossipBroadcast, GobEncode(c.hash, srcName, update.Encode())}
// FIXME a single blocked connection can stall us
for _, conn := range c.ourself.ConnectionsTo(nextHops) {
conn.(ProtocolSender).SendProtocolMsg(protocolMsg)
+2 -2
View File
@@ -377,8 +377,8 @@ func (router *Router) OnGossipUnicast(sender PeerName, msg []byte) error {
return fmt.Errorf("unexpected topology gossip unicast: %v", msg)
}
func (router *Router) OnGossipBroadcast(msg []byte) error {
return fmt.Errorf("unexpected topology gossip broadcast: %v", msg)
func (router *Router) OnGossipBroadcast(update []byte) (GossipData, error) {
return nil, fmt.Errorf("unexpected topology gossip broadcast: %v", update)
}
func (router *Router) Gossip() GossipData {