forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-17 01:13:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

whisper: messge bundling

Browse source
This commit is contained in:
Vlad 2017-12-13 15:37:54 +02:00
parent 3654aeaa4f
commit 9020bfbbe5
2 changed files with 28 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
whisper/whisperv6/peer.go
View file

@ -149,21 +149,26 @@ func (peer *Peer) expire() {
// broadcast iterates over the collection of envelopes and transmits yet unknown // broadcast iterates over the collection of envelopes and transmits yet unknown
// ones over the network. // ones over the network.
func (p *Peer) broadcast() error { func (p *Peer) broadcast() error {
var cnt int
envelopes := p.host.Envelopes() envelopes := p.host.Envelopes()
bundle := make([]*Envelope, 0, len(envelopes))
for _, envelope := range envelopes { for _, envelope := range envelopes {
if !p.marked(envelope) { if !p.marked(envelope) {
err := p2p.Send(p.ws, messagesCode, envelope) bundle = append(bundle, envelope)
if err != nil {
return err
} else {
p.mark(envelope)
cnt++
}
} }
} }
if cnt > 0 {
log.Trace("broadcast", "num. messages", cnt) // transmit the unknown batch (potentially empty)
if err := p2p.Send(p.ws, messagesCode, bundle); err != nil {
return err
}
// mark envelopes only if they were successfully sent
for _, e := range bundle {
p.mark(e)
}
if len(bundle) > 0 {
log.Trace("broadcast", "num. messages", len(bundle))
} }
return nil return nil
} }

24
whisper/whisperv6/whisper.go
View file

@ -515,18 +515,20 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
log.Warn("unxepected status message received", "peer", p.peer.ID()) log.Warn("unxepected status message received", "peer", p.peer.ID())
case messagesCode: case messagesCode:
// decode the contained envelopes // decode the contained envelopes
var envelope Envelope var envelopes []*Envelope
if err := packet.Decode(&envelope); err != nil { if err := packet.Decode(&envelopes); err != nil {
log.Warn("failed to decode envelope, peer will be disconnected", "peer", p.peer.ID(), "err", err) log.Warn("failed to decode envelopes, peer will be disconnected", "peer", p.peer.ID(), "err", err)
return errors.New("invalid envelope") return errors.New("invalid envelopes")
} }
cached, err := wh.add(&envelope) for _, env := range envelopes {
if err != nil { cached, err := wh.add(env)
log.Warn("bad envelope received, peer will be disconnected", "peer", p.peer.ID(), "err", err) if err != nil {
return errors.New("invalid envelope") log.Warn("bad envelope received, peer will be disconnected", "peer", p.peer.ID(), "err", err)
} return errors.New("invalid envelope")
if cached { }
p.mark(&envelope) if cached {
p.mark(env)
}
} }
case p2pCode: case p2pCode:
// peer-to-peer message, sent directly to peer bypassing PoW checks, etc. // peer-to-peer message, sent directly to peer bypassing PoW checks, etc.