code: func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) {
  	var msg *ReceivedMessage
  
  	fs.mutex.RLock()
  	defer fs.mutex.RUnlock()
  
  	candidates := fs.getWatchersByTopic(env.Topic)
  	for _, watcher := range candidates {
  		if p2pMessage && !watcher.AllowP2P {
  			log.Trace(fmt.Sprintf("msg [%x], filter [%s]: p2p messages are not allowed", env.Hash(), watcher.id))
  			continue
  		}
  
  		var match bool
  		if msg != nil {
  			match = watcher.MatchMessage(msg)
  		} else {
  			match = watcher.MatchEnvelope(env)
  			if match {
  				msg = env.Open(watcher)
  				if msg == nil {
  					log.Trace("processing message: failed to open", "message", env.Hash().Hex(), "filter", watcher.id)
  				}
  			} else {
  				log.Trace("processing message: does not match", "message", env.Hash().Hex(), "filter", watcher.id)
  			}
  		}
  
  		if match && msg != nil {
  			log.Trace("processing message: decrypted", "hash", env.Hash().Hex())
  			if watcher.Src == nil || IsPubKeyEqual(msg.Src, watcher.Src) {
  				watcher.Trigger(msg)
  			}
  		}
  	}
  }
docs: // NotifyWatchers notifies any filter that has declared interest
  // for the envelope's topic.