diff --git a/router/gossip.go b/router/gossip.go index 9f1a30a4..78c2706b 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -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) diff --git a/router/router.go b/router/router.go index 7ef3a9a2..756d39b7 100644 --- a/router/router.go +++ b/router/router.go @@ -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 {