pubsub.go
func (p *PubSub) pushMsg(vals []*topicVal, src peer.ID, msg *Message) {
	// reject messages from blacklisted peers
	if p.blacklist.Contains(src) {
		log.Warningf("dropping message from blacklisted peer %s", src)
		return
	}

	// even if they are forwarded by good peers
	if p.blacklist.Contains(msg.GetFrom()) {
		log.Warningf("dropping message from blacklisted source %s", src)
		return
	}

	// reject unsigned messages when strict before we even process the id
	if p.signStrict && msg.Signature == nil {
		log.Debugf("dropping unsigned message from %s", src)
		return
	}

	// have we already seen and validated this message?
	id := msgID(msg.Message)
	if p.seenMessage(id) {
		return
	}

	if len(vals) > 0 || msg.Signature != nil {
		// validation is asynchronous and globally throttled with the throttleValidate semaphore.
		// the purpose of the global throttle is to bound the goncurrency possible from incoming
		// network traffic; each validator also has an individual throttle to preclude
		// slow (or faulty) validators from starving other topics; see validate below.
		select {
		case p.validateThrottle <- struct{}{}:
			go func() {
				p.validate(vals, src, msg)
				<-p.validateThrottle
			}()
		default:
			log.Warningf("message validation throttled; dropping message from %s", src)
		}
		return
	}

	p.publishMessage(src, msg.Message)
}
func (p *PubSub) pushMsg(vals []*topicVal, src peer.ID, msg *Message) {
	// reject messages from blacklisted peers
	if p.blacklist.Contains(src) {
		log.Warningf("dropping message from blacklisted peer %s", src)
		return
	}

	// even if they are forwarded by good peers
	if p.blacklist.Contains(msg.GetFrom()) {
		log.Warningf("dropping message from blacklisted source %s", src)
		return
	}

	// reject unsigned messages when strict before we even process the id
	if p.signStrict && msg.Signature == nil {
		log.Debugf("dropping unsigned message from %s", src)
		return
	}

	// have we already seen and validated this message?
	id := msgID(msg.Message)
	if p.seenMessage(id) {
		return
	}

	if len(vals) > 0 || msg.Signature != nil {
		// validation is asynchronous and globally throttled with the throttleValidate semaphore.
		// the purpose of the global throttle is to bound the goncurrency possible from incoming
		// network traffic; each validator also has an individual throttle to preclude
		// slow (or faulty) validators from starving other topics; see validate below.
		select {
		case p.validateThrottle <- struct{}{}:
			go func() {
				p.validate(vals, src, msg)
				<-p.validateThrottle
			}()
		default:
			log.Warningf("message validation throttled; dropping message from %s", src)
		}
		return
	}

	p.publishMessage(src, msg.Message)
}
// pushMsg pushes a message performing validation as necessary