grpc/streambuffer/streambuffer.go
func (s *Stream) Run() (err error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	defer func() {
		if err != nil {
			if grpc.Code(err) == codes.Canceled {
				s.log.Debug("streambuffer: context canceled")
				err = context.Canceled
				return
			}
			if grpc.Code(err) == codes.DeadlineExceeded {
				s.log.Debug("streambuffer: context deadline exceeded")
				err = context.DeadlineExceeded
				return
			}
		}
	}()

	stream, err := s.setupFunc()
	if err != nil {
		s.log.WithError(err).Debug("streambuffer: setup returned error")
		return err
	}

	recvErr := make(chan error)
	defer func() {
		go func() { // empty the recvErr channel on return
			<-recvErr
		}()
	}()

	go func() {
		for {
			var r interface{}
			if s.recvFunc != nil {
				r = s.recvFunc()
			} else {
				r = new(empty.Empty) // Generic proto message if not interested in received values
			}
			err := stream.RecvMsg(r)
			if err != nil {
				s.log.WithError(err).Debug("streambuffer: error from stream.RecvMsg")
				recvErr <- err
				close(recvErr)
				return
			}
			if s.recvFunc != nil {
				s.recvMsg(r)
			}
		}
	}()

	defer stream.CloseSend()

	for {
		select {
		case err := <-recvErr:
			return err
		case <-stream.Context().Done():
			s.log.WithError(stream.Context().Err()).Debug("streambuffer: context done")
			return stream.Context().Err()
		case msg := <-s.sendBuffer:
			if err = stream.SendMsg(msg); err != nil {
				s.log.WithError(err).Debug("streambuffer: error from stream.SendMsg")
				return err
			}
			atomic.AddUint64(&s.sent, 1)
		}
	}
}
func (s *Stream) Run() (err error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	defer func() {
		if err != nil {
			if grpc.Code(err) == codes.Canceled {
				s.log.Debug("streambuffer: context canceled")
				err = context.Canceled
				return
			}
			if grpc.Code(err) == codes.DeadlineExceeded {
				s.log.Debug("streambuffer: context deadline exceeded")
				err = context.DeadlineExceeded
				return
			}
		}
	}()

	stream, err := s.setupFunc()
	if err != nil {
		s.log.WithError(err).Debug("streambuffer: setup returned error")
		return err
	}

	recvErr := make(chan error)
	defer func() {
		go func() { // empty the recvErr channel on return
			<-recvErr
		}()
	}()

	go func() {
		for {
			var r interface{}
			if s.recvFunc != nil {
				r = s.recvFunc()
			} else {
				r = new(empty.Empty) // Generic proto message if not interested in received values
			}
			err := stream.RecvMsg(r)
			if err != nil {
				s.log.WithError(err).Debug("streambuffer: error from stream.RecvMsg")
				recvErr <- err
				close(recvErr)
				return
			}
			if s.recvFunc != nil {
				s.recvMsg(r)
			}
		}
	}()

	defer stream.CloseSend()

	for {
		select {
		case err := <-recvErr:
			return err
		case <-stream.Context().Done():
			s.log.WithError(stream.Context().Err()).Debug("streambuffer: context done")
			return stream.Context().Err()
		case msg := <-s.sendBuffer:
			if err = stream.SendMsg(msg); err != nil {
				s.log.WithError(err).Debug("streambuffer: error from stream.SendMsg")
				return err
			}
			atomic.AddUint64(&s.sent, 1)
		}
	}
}
// Run the stream.
//
// This calls the underlying grpc.ClientStreams methods to send and receive messages over the stream.
// Run returns the error returned by any of those functions, or context.Canceled if the context is canceled.