code: func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer {
  	p := &Peer{
  		Peer:         peer,
  		pq:           pq.New(int(PriorityQueue), PriorityQueueCap),
  		streamer:     streamer,
  		servers:      make(map[Stream]*server),
  		clients:      make(map[Stream]*client),
  		clientParams: make(map[Stream]*clientParams),
  		quit:         make(chan struct{}),
  	}
  	ctx, cancel := context.WithCancel(context.Background())
  	go p.pq.Run(ctx, func(i interface{}) {
  		wmsg := i.(WrappedPriorityMsg)
  		err := p.Send(wmsg.Context, wmsg.Msg)
  		if err != nil {
  			log.Error("Message send error, dropping peer", "peer", p.ID(), "err", err)
  			p.Drop(err)
  		}
  	})
  
  	// basic monitoring for pq contention
  	go func(pq *pq.PriorityQueue) {
  		ticker := time.NewTicker(5 * time.Second)
  		defer ticker.Stop()
  		for {
  			select {
  			case <-ticker.C:
  				var lenMaxi int
  				var capMaxi int
  				for k := range pq.Queues {
  					if lenMaxi < len(pq.Queues[k]) {
  						lenMaxi = len(pq.Queues[k])
  					}
  
  					if capMaxi < cap(pq.Queues[k]) {
  						capMaxi = cap(pq.Queues[k])
  					}
  				}
  
  				metrics.GetOrRegisterGauge(fmt.Sprintf("pq_len_%s", p.ID().TerminalString()), nil).Update(int64(lenMaxi))
  				metrics.GetOrRegisterGauge(fmt.Sprintf("pq_cap_%s", p.ID().TerminalString()), nil).Update(int64(capMaxi))
  			case <-p.quit:
  				return
  			}
  		}
  	}(p.pq)
  
  	go func() {
  		<-p.quit
  
  		cancel()
  	}()
  	return p
  }
docs: // NewPeer is the constructor for Peer