whisper: refactoring of message queuing

whisper/whisperv5/doc.go

@@ -40,11 +40,11 @@ const (
 	ProtocolVersionStr = "5.0"
 	ProtocolName       = "shh"
 
-	statusCode           = 0
+	statusCode           = 0 // used by whisper protocol
-	messagesCode         = 1
+	messagesCode         = 1 // normal whisper message
-	p2pCode              = 2
+	p2pCode              = 2 // peer-to-peer message (to be consumed by the peer, but not forwarded any futher)
-	mailRequestCode      = 3
+	p2pRequestCode       = 3 // peer-to-peer message, used by Dapp protocol
-	NumberOfMessageCodes = 32
+	NumberOfMessageCodes = 64
 
 	paddingMask   = byte(3)
 	signatureFlag = byte(4)

whisper/whisperv5/filter.go

@@ -76,11 +76,11 @@ func (fs *Filters) Get(i uint32) *Filter {
 	return fs.watchers[i]
 }
 
-func (fs *Filters) NotifyWatchers(env *Envelope, messageCode uint64) {
+func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) {
 	fs.mutex.RLock()
 	var msg *ReceivedMessage
 	for j, watcher := range fs.watchers {
-		if messageCode == p2pCode && !watcher.AcceptP2P {
+		if p2pMessage && !watcher.AcceptP2P {
 			glog.V(9).Infof("msg [%x], filter [%d]: p2p messages are not allowed \n", env.Hash(), j)
 			continue
 		}

whisper/whisperv5/filter_test.go

@@ -540,7 +540,7 @@ func TestWatchers(t *testing.T) {
 	}
 
 	for i = 0; i < NumMessages; i++ {
-		filters.NotifyWatchers(envelopes[i], messagesCode)
+		filters.NotifyWatchers(envelopes[i], false)
 	}
 
 	var total int
@@ -593,7 +593,7 @@ func TestWatchers(t *testing.T) {
 	}
 
 	for i = 0; i < NumMessages; i++ {
-		filters.NotifyWatchers(envelopes[i], messagesCode)
+		filters.NotifyWatchers(envelopes[i], false)
 	}
 
 	for i = 0; i < NumFilters; i++ {
@@ -629,7 +629,7 @@ func TestWatchers(t *testing.T) {
 	// test AcceptP2P
 
 	total = 0
-	filters.NotifyWatchers(envelopes[0], p2pCode)
+	filters.NotifyWatchers(envelopes[0], true)
 
 	for i = 0; i < NumFilters; i++ {
 		mail = tst[i].f.Retrieve()
@@ -646,7 +646,7 @@ func TestWatchers(t *testing.T) {
 	}
 	f.AcceptP2P = true
 	total = 0
-	filters.NotifyWatchers(envelopes[0], p2pCode)
+	filters.NotifyWatchers(envelopes[0], true)
 
 	for i = 0; i < NumFilters; i++ {
 		mail = tst[i].f.Retrieve()

whisper/whisperv5/whisper.go

@@ -36,11 +36,6 @@ import (
 	set "gopkg.in/fatih/set.v0"
 )
 
-type QueueItem struct {
-	hash common.Hash
-	code uint64
-}
-
 // Whisper represents a dark communication interface through the Ethereum
 // network, using its very own P2P communication layer.
 type Whisper struct {
@@ -61,7 +56,8 @@ type Whisper struct {
 
 	mailServer MailServer
 
-	messageQueue chan QueueItem
+	messageQueue chan *Envelope
+	p2pMsgQueue  chan *Envelope
 	quit         chan struct{}
 
 	overflow bool
@@ -79,7 +75,8 @@ func NewWhisper(server MailServer) *Whisper {
 		expirations:  make(map[uint32]*set.SetNonTS),
 		peers:        make(map[*Peer]struct{}),
 		mailServer:   server,
-		messageQueue: make(chan QueueItem, messageQueueLimit),
+		messageQueue: make(chan *Envelope, messageQueueLimit),
+		p2pMsgQueue:  make(chan *Envelope, messageQueueLimit),
 		quit:         make(chan struct{}),
 	}
 	whisper.filters = NewFilters(whisper)
@@ -134,7 +131,7 @@ func (w *Whisper) RequestHistoricMessages(peerID []byte, data []byte) error {
 		return err
 	}
 	p.trusted = true
-	return p2p.Send(p.ws, mailRequestCode, data)
+	return p2p.Send(p.ws, p2pRequestCode, data)
 }
 
 func (w *Whisper) SendP2PMessage(peerID []byte, envelope *Envelope) error {
@@ -369,7 +366,7 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
 					wh.postEvent(envelope, p2pCode)
 				}
 			}
-		case mailRequestCode:
+		case p2pRequestCode:
 			// Must be processed if mail server is implemented. Otherwise ignore.
 			if wh.mailServer != nil {
 				s := rlp.NewStream(packet.Payload, uint64(packet.Size))
@@ -467,9 +464,12 @@ func (w *Whisper) postEvent(envelope *Envelope, messageCode uint64) {
 	// currently supported version, we can not decrypt it,
 	// and therefore just ignore this message
 	if envelope.Ver() <= EnvelopeVersion {
-		w.checkOverflow()
+		if messagesCode == messageCode {
-		i := QueueItem{hash: envelope.Hash(), code: messageCode}
+			w.checkOverflow()
-		w.messageQueue <- i
+			w.messageQueue <- envelope
+		} else {
+			w.p2pMsgQueue <- envelope
+		}
 	}
 }
 
@@ -491,18 +491,17 @@ func (w *Whisper) checkOverflow() {
 
 // processQueue delivers the messages to the watchers during the lifetime of the whisper node.
 func (w *Whisper) processQueue() {
+	var e *Envelope
 	for {
 		select {
 		case <-w.quit:
 			return
 
-		case i := <-w.messageQueue:
+		case e = <-w.messageQueue:
-			w.poolMu.Lock()
+			w.filters.NotifyWatchers(e, false)
-			e := w.envelopes[i.hash]
+
-			w.poolMu.Unlock()
+		case e = <-w.p2pMsgQueue:
-			if e != nil {
+			w.filters.NotifyWatchers(e, true)
-				w.filters.NotifyWatchers(e, i.code)
-			}
 		}
 	}
 }