func (s *Stream) SetTrailer(md metadata.MD) error {
	if md.Len() == 0 {
		return nil
	}
	if s.getState() == streamDone {
		return ErrIllegalHeaderWrite
	}
	s.hdrMu.Lock()
	s.trailer = metadata.Join(s.trailer, md)
	s.hdrMu.Unlock()
	return nil
}
SetTrailer sets the trailer metadata which will be sent with the RPC status
by the server. This can be called multiple times. Server side only.
This should not be called parallel to other data writes.
package transport
import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"net"
	"sync"
	"sync/atomic"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/keepalive"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/resolver"
	"google.golang.org/grpc/stats"
	"google.golang.org/grpc/status"
	"google.golang.org/grpc/tap"
)
const logLevel = 2
type bufferPool struct {
	pool sync.Pool
}
func newBufferPool() *bufferPool {
	return &bufferPool{
		pool: sync.Pool{
			New: func() interface{} {
				return new(bytes.Buffer)
			},
		},
	}
}
func (p *bufferPool) get() *bytes.Buffer {
	return p.pool.Get().(*bytes.Buffer)
}
func (p *bufferPool) put(b *bytes.Buffer) {
	p.pool.Put(b)
}
type recvMsg struct {
	buffer *bytes.Buffer
	err error
}
type recvBuffer struct {
	c       chan recvMsg
	mu      sync.Mutex
	backlog []recvMsg
	err     error
}
func newRecvBuffer() *recvBuffer {
	b := &recvBuffer{
		c: make(chan recvMsg, 1),
	}
	return b
}
func (b *recvBuffer) put(r recvMsg) {
	b.mu.Lock()
	if b.err != nil {
		b.mu.Unlock()
		return
	}
	b.err = r.err
	if len(b.backlog) == 0 {
		select {
		case b.c <- r:
			b.mu.Unlock()
			return
		default:
		}
	}
	b.backlog = append(b.backlog, r)
	b.mu.Unlock()
}
func (b *recvBuffer) load() {
	b.mu.Lock()
	if len(b.backlog) > 0 {
		select {
		case b.c <- b.backlog[0]:
			b.backlog[0] = recvMsg{}
			b.backlog = b.backlog[1:]
		default:
		}
	}
	b.mu.Unlock()
}
func (b *recvBuffer) get() <-chan recvMsg {
	return b.c
}
type recvBufferReader struct {
	closeStream func(error)  
	ctx         context.Context
	ctxDone     <-chan struct{}  
	recv        *recvBuffer
	last        *bytes.Buffer  
	err         error
	freeBuffer  func(*bytes.Buffer)
}
func (r *recvBufferReader) Read(p []byte) (n int, err error) {
	if r.err != nil {
		return 0, r.err
	}
	if r.last != nil {
		copied, _ := r.last.Read(p)
		if r.last.Len() == 0 {
			r.freeBuffer(r.last)
			r.last = nil
		}
		return copied, nil
	}
	if r.closeStream != nil {
		n, r.err = r.readClient(p)
	} else {
		n, r.err = r.read(p)
	}
	return n, r.err
}
func (r *recvBufferReader) read(p []byte) (n int, err error) {
	select {
	case <-r.ctxDone:
		return 0, ContextErr(r.ctx.Err())
	case m := <-r.recv.get():
		return r.readAdditional(m, p)
	}
}
func (r *recvBufferReader) readClient(p []byte) (n int, err error) {
	select {
	case <-r.ctxDone:
		r.closeStream(ContextErr(r.ctx.Err()))
		m := <-r.recv.get()
		return r.readAdditional(m, p)
	case m := <-r.recv.get():
		return r.readAdditional(m, p)
	}
}
func (r *recvBufferReader) readAdditional(m recvMsg, p []byte) (n int, err error) {
	r.recv.load()
	if m.err != nil {
		return 0, m.err
	}
	copied, _ := m.buffer.Read(p)
	if m.buffer.Len() == 0 {
		r.freeBuffer(m.buffer)
		r.last = nil
	} else {
		r.last = m.buffer
	}
	return copied, nil
}
type streamState uint32
const (
	streamActive    streamState = iota
	streamWriteDone              
	streamReadDone               
	streamDone                   
)
type Stream struct {
	id           uint32
	st           ServerTransport     
	ct           *http2Client        
	ctx          context.Context     
	cancel       context.CancelFunc  
	done         chan struct{}       
	ctxDone      <-chan struct{}     
	method       string              
	recvCompress string
	sendCompress string
	buf          *recvBuffer
	trReader     io.Reader
	fc           *inFlow
	wq           *writeQuota
	requestRead func(int)
	headerChan       chan struct{}  
	headerChanClosed uint32         
	headerValid bool
	hdrMu sync.Mutex
	header  metadata.MD
	trailer metadata.MD  
	noHeaders bool  
	headerSent uint32
	state streamState
	status *status.Status
	bytesReceived uint32  
	unprocessed   uint32  
	contentSubtype string
}
func (s *Stream) isHeaderSent() bool {
	return atomic.LoadUint32(&s.headerSent) == 1
}
func (s *Stream) updateHeaderSent() bool {
	return atomic.SwapUint32(&s.headerSent, 1) == 1
}
func (s *Stream) swapState(st streamState) streamState {
	return streamState(atomic.SwapUint32((*uint32)(&s.state), uint32(st)))
}
func (s *Stream) compareAndSwapState(oldState, newState streamState) bool {
	return atomic.CompareAndSwapUint32((*uint32)(&s.state), uint32(oldState), uint32(newState))
}
func (s *Stream) getState() streamState {
	return streamState(atomic.LoadUint32((*uint32)(&s.state)))
}
func (s *Stream) waitOnHeader() {
	if s.headerChan == nil {
		return
	}
	select {
	case <-s.ctx.Done():
		s.ct.CloseStream(s, ContextErr(s.ctx.Err()))
		<-s.headerChan
	case <-s.headerChan:
	}
}
func (s *Stream) RecvCompress() string {
	s.waitOnHeader()
	return s.recvCompress
}
func (s *Stream) SetSendCompress(str string) {
	s.sendCompress = str
}
func (s *Stream) Done() <-chan struct{} {
	return s.done
}
func (s *Stream) Header() (metadata.MD, error) {
	if s.headerChan == nil {
		return s.header.Copy(), nil
	}
	s.waitOnHeader()
	if !s.headerValid {
		return nil, s.status.Err()
	}
	return s.header.Copy(), nil
}
func (s *Stream) TrailersOnly() bool {
	s.waitOnHeader()
	return s.noHeaders
}
func (s *Stream) Trailer() metadata.MD {
	c := s.trailer.Copy()
	return c
}
func (s *Stream) ContentSubtype() string {
	return s.contentSubtype
}
func (s *Stream) Context() context.Context {
	return s.ctx
}
func (s *Stream) Method() string {
	return s.method
}
func (s *Stream) Status() *status.Status {
	return s.status
}
func (s *Stream) SetHeader(md metadata.MD) error {
	if md.Len() == 0 {
		return nil
	}
	if s.isHeaderSent() || s.getState() == streamDone {
		return ErrIllegalHeaderWrite
	}
	s.hdrMu.Lock()
	s.header = metadata.Join(s.header, md)
	s.hdrMu.Unlock()
	return nil
}
func (s *Stream) SendHeader(md metadata.MD) error {
	return s.st.WriteHeader(s, md)
}