grpc/streambuffer/streambuffer.go
func (s *Stream) recvMsg(msg interface{}) {
	s.mu.RLock()
	if s.recvBuffer == nil {
		s.mu.RUnlock()
		return
	}
	defer s.mu.RUnlock()
	select {
	case s.recvBuffer <- msg: // normal flow if the channel is not blocked
	default:
		s.log.Debug("streambuffer: dropping received message")
		atomic.AddUint64(&s.dropped, 1)
		<-s.recvBuffer // drop oldest and try again (if application temporarily unavailable)
		select {
		case s.recvBuffer <- msg:
		default: // drop newest (too many cuncurrent recvMsg)
			atomic.AddUint64(&s.dropped, 1)
			s.log.Debug("streambuffer: dropping received message")
		}
	}
}
func (s *Stream) recvMsg(msg interface{}) {
	s.mu.RLock()
	if s.recvBuffer == nil {
		s.mu.RUnlock()
		return
	}
	defer s.mu.RUnlock()
	select {
	case s.recvBuffer <- msg: // normal flow if the channel is not blocked
	default:
		s.log.Debug("streambuffer: dropping received message")
		atomic.AddUint64(&s.dropped, 1)
		<-s.recvBuffer // drop oldest and try again (if application temporarily unavailable)
		select {
		case s.recvBuffer <- msg:
		default: // drop newest (too many cuncurrent recvMsg)
			atomic.AddUint64(&s.dropped, 1)
			s.log.Debug("streambuffer: dropping received message")
		}
	}
}
// recvMsg receives a message (possibly dropping a message on full buffers)