whisper/whisperv6: message bundling (#15666)

Changed the communication protocol for ordinary message,
according to EIP 627. Messages will be send in bundles, i.e.
array of messages will be sent instead of single message.

whisper/whisperv6/peer.go

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

whisper/whisperv6/whisper.go

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